"""
KiloCodeAdapter — connects to a local Kilo Code server via HTTP+SSE.

Kilo Code (github.com/Kilo-Org/kilocode, MIT) is an open-source agentic
coding platform ranked #2 on OpenRouter by weekly token volume (~143B/week).
This adapter uses the HTTP API exposed by `kilo serve`.

Architecture:
  `kilo serve --port 0` starts a Hono HTTP server on a random port.
  The assigned port is printed to stdout: "Listening on http://127.0.0.1:{port}".
  All real-time events flow through GET /event (global SSE bus, one connection
  per instance shared across all sessions).

API:
  Auth:  KILO_SERVER_PASSWORD env var → x-kilo-password header
  Scope: x-kilo-directory header pins every request to a project directory

  POST /session                        create session → {id, ...}
  POST /session/{id}/prompt_async      send message → 204 (fire-and-forget)
  POST /session/{id}/abort             stop current turn → 200
  POST /permission/{requestID}/reply   {reply: "once"|"always"|"reject"}
  GET  /event                          SSE bus — all events for this instance

SSE events we handle (each has shape {id, type, properties: {...}}):
  message.part.delta    {sessionID, partID, field, delta}
                         field="text" → streaming TextBlock or ThinkingBlock
  message.part.updated  {sessionID, part: {type, id, ...}}  (SyncEvent on bus)
                         type="tool-invocation" → ToolCallBlock state transitions
                         type="text"            → finalize TextBlock
                         type="reasoning"       → finalize ThinkingBlock
  permission.asked      full Permission.Request → PermissionRequestBlock
  session.idle          {sessionID} → SessionEndBlock(completed)   [deprecated but still emitted]
  session.status        {sessionID, status: {type}} → idle → SessionEndBlock
  session.error         {sessionID, error} → ErrorBlock + SessionEndBlock
  server.heartbeat      ignore
  server.connected      ignore
"""
from __future__ import annotations

import asyncio
import base64
import json
import logging
import os
import re
import secrets
import shutil
import sqlite3
import subprocess
import uuid
from typing import AsyncIterator

log = logging.getLogger(__name__)

from ..platform_paths import user_data_dir as _data_dir

_KILO_DB = _data_dir("kilo") / "kilo.db"

# Title generated by Kilo for brand-new sessions that never had a real exchange.
_NEW_SESSION_RE = re.compile(r"^New session\s*[-–]", re.IGNORECASE)


def _scan_kilo_sessions(limit: int = 20) -> list["ExternalSession"]:
    """Read recent sessions from ~/.local/share/kilo/kilo.db."""
    from .base import ExternalSession

    if not _KILO_DB.exists():
        return []

    try:
        con = sqlite3.connect(f"file:{_KILO_DB}?mode=ro", uri=True,
                              check_same_thread=False)
        con.row_factory = sqlite3.Row
    except Exception as e:
        log.debug("kilo scan: cannot open DB: %s", e)
        return []

    try:
        rows = con.execute("""
            SELECT
                s.id,
                s.directory,
                s.title,
                s.model,
                datetime(s.time_created / 1000, 'unixepoch') AS started_at,
                (
                    SELECT json_extract(p.data, '$.text')
                    FROM   part p
                    JOIN   message m ON p.message_id = m.id
                    WHERE  m.session_id = s.id
                      AND  json_extract(m.data,  '$.role')  = 'user'
                      AND  json_extract(p.data,  '$.type')  = 'text'
                    ORDER BY m.time_created, p.time_created
                    LIMIT 1
                ) AS first_prompt
            FROM  session s
            WHERE s.time_archived IS NULL
              AND EXISTS (
                    SELECT 1 FROM message m
                    WHERE  m.session_id = s.id
                      AND  json_extract(m.data, '$.role') = 'user'
              )
            ORDER BY s.time_updated DESC
            LIMIT ?
        """, (limit,)).fetchall()
    except Exception as e:
        log.debug("kilo scan: query failed: %s", e)
        return []
    finally:
        con.close()

    # is_active: a kilo process is currently running (single server, not per-session)
    kilo_running = _kilo_process_running()

    result: list[ExternalSession] = []
    for row in rows:
        title = row["title"] or ""
        first_prompt = row["first_prompt"] or ""
        # Use the AI-generated title as the last_prompt when it's meaningful;
        # fall back to the raw first message text otherwise.
        if title and not _NEW_SESSION_RE.match(title):
            last_prompt = title
        else:
            last_prompt = first_prompt

        started_at = row["started_at"] or ""
        if started_at:
            # sqlite datetime() returns "YYYY-MM-DD HH:MM:SS" — normalise to ISO-8601
            started_at = started_at.replace(" ", "T") + "Z"

        result.append(ExternalSession(
            session_id=row["id"],
            agent_id="kilo",
            cwd=row["directory"] or "",
            last_prompt=last_prompt,
            started_at=started_at,
            is_active=kilo_running,
        ))

    return result


def _kilo_process_running() -> bool:
    """Return True if any kilo process is currently running."""
    try:
        out = subprocess.check_output(
            ["pgrep", "-x", "kilo"], text=True, stderr=subprocess.DEVNULL
        )
        return bool(out.strip())
    except Exception:
        return False


# A Kilo session in `cwd` updated within this window counts as actively driven
# elsewhere (the editor/another server). Beacon then parks rather than co-writing.
_KILO_ACTIVE_WINDOW_MS = 60_000


def _kilo_sessions_in_cwd(cwd: str) -> list[tuple[str, int]]:
    """[(session_id, time_updated_ms)] for non-archived Kilo sessions whose directory
    resolves to `cwd`, newest first. Used by the session-handoff hooks."""
    if not _KILO_DB.exists() or not cwd:
        return []
    target = os.path.realpath(os.path.expanduser(cwd))
    try:
        con = sqlite3.connect(f"file:{_KILO_DB}?mode=ro", uri=True, check_same_thread=False)
        con.row_factory = sqlite3.Row
    except Exception:
        return []
    try:
        rows = con.execute("""
            SELECT id, directory, time_updated FROM session
            WHERE time_archived IS NULL AND directory IS NOT NULL
            ORDER BY time_updated DESC
        """).fetchall()
    except Exception:
        return []
    finally:
        con.close()
    out: list[tuple[str, int]] = []
    for r in rows:
        try:
            if os.path.realpath(r["directory"]) == target:
                out.append((r["id"], int(r["time_updated"] or 0)))
        except Exception:
            continue
    return out

import httpx

from ..attachments import Attachment, build_text_with_inline
from ..blocks import (
    AgentInfoBlock,
    AudioHintBlock,
    Block,
    ErrorBlock,
    InputRequestBlock,
    PermissionRequestBlock,
    SessionEndBlock,
    SessionStatusHintBlock,
    SuggestionsBlock,
    SystemStatusBlock,
    TextBlock,
    ThinkingBlock,
    ToolCallBlock,
    tool_result_or_large,
)
from .base import (BaseAdapter, DetectedAgent, ExternalSession, split_markdown,
                   agent_env, tool_label_subject, context_window_for_model, probe_binary)

_DEFAULT_PORT = 4096

_COMPACT_PROMPT = (
    "Condense the conversation history. Use this structure:\n\n"
    "## What Was Done\n"
    "Bullet points of completed changes, with exact file paths.\n\n"
    "## Current State\n"
    "Where things stand, any in-progress work.\n\n"
    "## Important Context\n"
    "Key decisions, constraints, or facts needed to continue."
)


def find_kilo_binary() -> str | None:
    return shutil.which("kilo")


def find_kilo_port() -> int | None:
    """Return the port if a kilo serve instance is already running at the default port."""
    try:
        resp = httpx.get(f"http://127.0.0.1:{_DEFAULT_PORT}/health", timeout=1)
        if resp.status_code < 500:
            return _DEFAULT_PORT
    except Exception:
        pass
    return None


def _read_kilo_model_config() -> dict:
    """Return {current_model, recent_models, favorite_models} from Kilo config files."""
    result: dict = {"current_model": "", "recent_models": [], "favorite_models": []}
    from ..platform_paths import user_config_dir, user_state_dir
    cli_cfg = user_config_dir("kilo") / "kilo.json"
    if cli_cfg.exists():
        try:
            data = json.loads(cli_cfg.read_text())
            result["current_model"] = data.get("model", "")
        except Exception:
            pass
    state_cfg = user_state_dir("kilo") / "model.json"
    if state_cfg.exists():
        try:
            data = json.loads(state_cfg.read_text())

            def _fmt(entry: dict) -> str:
                return f"{entry['providerID']}/{entry['modelID']}"
            result["recent_models"] = [_fmt(e) for e in data.get("recent", []) if "providerID" in e]
            result["favorite_models"] = [_fmt(e) for e in data.get("favorite", []) if "providerID" in e]
            if not result["current_model"]:
                code = data.get("model", {}).get("code", {})
                if code.get("providerID"):
                    result["current_model"] = _fmt(code)
        except Exception:
            pass
    return result


class KiloCodeAdapter(BaseAdapter):
    cap_images = True
    cap_documents = False
    cap_thinking = True
    cap_permissions = True
    cap_subagents = True
    cap_interrupt = True
    # Beacon sentinel tools: kilo (an OpenCode-family server) loads the globally-registered
    # beacon MCP from ~/.config/kilo/kilo.json (written by setup_integrations._setup_kilo) and,
    # like opencode, passes the `kilo serve` spawn env through to the local MCP child — so
    # TERMINULL_SENTINEL_PORT/TERMINULL_SESSION_ID stamped into the server env route the call back to
    # this session. MEASURED end-to-end (P4).
    cap_sentinel = True
    provides_models = True
    provides_usage = True
    agent_id = "kilo"
    display_name = "Kilo Code"
    beta = False              # feature-complete (model + todo + sentinel, live-verified) — promoted out of Beta

    @classmethod
    def detect(cls) -> "DetectedAgent | None":
        binary = shutil.which("kilo")
        if not binary:
            return None
        return DetectedAgent(
            id="kilo", name="Kilo Code", binary=binary,
            available=probe_binary(binary),
            extra=_read_kilo_model_config(),
        )

    def __init__(
        self,
        base_url: str | None = None,
        cwd: str | None = None,
        binary: str | None = None,
        compact_threshold: float = 0.5,
        resume_kilo_session_id: str | None = None,
    ) -> None:
        self._base_url = base_url        # resolved after spawn if None
        self._cwd = cwd
        self._binary = binary or find_kilo_binary() or "kilo"
        self._compact_threshold = compact_threshold  # 0.0 = disabled
        self._resume_kilo_session_id = resume_kilo_session_id
        self._proc: asyncio.subprocess.Process | None = None
        self._kilo_session_id: str = ""
        self._password: str | None = None
        self._send_queue: asyncio.Queue[dict] = asyncio.Queue()

        # Streaming state per part
        self._part_types: dict[str, str] = {}          # partID → "text" | "reasoning" | "tool-invocation"
        self._text_parts: dict[str, tuple[str, str]] = {}  # partID → (block_id, accumulated_text)
        self._thinking_parts: dict[str, str] = {}      # partID → accumulated_text
        self._pending_tools: dict[str, str] = {}       # toolCallId → block_id
        self._pending_questions: dict[str, list[dict]] = {}  # question_id → questions[]
        self._pending_network: set[str] = set()              # network recovery request IDs
        self._fetch_store: dict[str, str] = {}
        self._stopping: bool = False

        # Per-turn usage counters — reset at turn start, populated from message.updated
        self._turn_input_tokens: int = 0
        # Cached portion of the last prompt (cache read + write). The fresh `input` alone
        # wildly understates context occupancy once prompt caching kicks in, so "% full"
        # and the compaction threshold must count cache too (mirrors claude_code).
        self._turn_cache_tokens: int = 0
        self._turn_output_tokens: int = 0
        self._turn_cost_usd: float = 0.0

        # Cumulative usage across all turns this session
        self._total_input_tokens: int = 0
        self._total_output_tokens: int = 0
        self._total_cost_usd: float = 0.0
        self._total_turns: int = 0
        self._model: str = ""
        self._context_window: int = 0

    # ── Session discovery ─────────────────────────────────────────────────────

    @classmethod
    def scan_external_sessions(cls) -> list[ExternalSession]:
        return _scan_kilo_sessions()

    @classmethod
    def read_session_condensed(cls, ext_sid: str, max_turns: int = 20) -> str:
        """Return last `max_turns` user+assistant exchanges from kilo.db as condensed text."""
        if not _KILO_DB.exists():
            return ""
        try:
            con = sqlite3.connect(f"file:{_KILO_DB}?mode=ro", uri=True, check_same_thread=False)
            con.row_factory = sqlite3.Row
            rows = con.execute("""
                SELECT json_extract(m.data, '$.role') AS role,
                       json_extract(p.data, '$.text') AS text
                FROM   part p
                JOIN   message m ON p.message_id = m.id
                WHERE  m.session_id = ?
                  AND  json_extract(p.data, '$.type') = 'text'
                  AND  json_extract(m.data, '$.role') IN ('user', 'assistant')
                ORDER BY m.time_created, p.time_created
            """, (ext_sid,)).fetchall()
            con.close()
        except Exception:
            return ""
        turns = [f"{r['role']}: {(r['text'] or '')[:400]}" for r in rows if r['text']]
        return "\n\n".join(turns[-(max_turns * 2):])

    @classmethod
    def has_changes_since(cls, ts: float) -> bool:
        """True if kilo.db has been modified since `ts`."""
        try:
            return _KILO_DB.exists() and _KILO_DB.stat().st_mtime > ts
        except OSError:
            return False

    @classmethod
    def is_cwd_externally_owned(cls, cwd: str, sid: str = "") -> bool:
        # A kilo process is up AND a session in this cwd was written very recently
        # (editor / other server actively driving it). Recency avoids treating every
        # cwd-with-history as owned just because the editor is open. docs/session-handoff.md
        if not _kilo_process_running():
            return False
        import time as _t
        now_ms = _t.time() * 1000
        return any((now_ms - ts) < _KILO_ACTIVE_WINDOW_MS for _id, ts in _kilo_sessions_in_cwd(cwd))

    @classmethod
    def latest_sid_for_cwd(cls, cwd: str) -> str | None:
        sessions = _kilo_sessions_in_cwd(cwd)
        return sessions[0][0] if sessions else None

    @classmethod
    def make_transcript_mirror(cls, cwd: str, sid: str, *, from_end: bool = False):
        from ..transcript_mirror import make_kilo_mirror
        return make_kilo_mirror(sid, from_end=from_end)

    # ── Model discovery ───────────────────────────────────────────────────────

    @classmethod
    async def list_models(cls, extra: dict | None = None) -> list[dict]:
        extra = extra or {}
        seen: set[str] = set()
        models: list[dict] = []

        def _add(model_str: str, detail: str = "") -> None:
            if not model_str or model_str in seen:
                return
            seen.add(model_str)
            # Format: "providerID/modelID" — label is the last path component
            parts = model_str.split("/", 1)
            provider = parts[0] if len(parts) > 1 else ""
            model_path = parts[1] if len(parts) > 1 else model_str
            label = model_path.rsplit("/", 1)[-1]
            models.append({
                "id": model_str,
                "label": label,
                "detail": detail or (f"via {provider}" if provider else ""),
            })

        current = extra.get("current_model", "")
        _add(current, "Current model")
        for m in extra.get("favorite_models", []):
            _add(m, "Favorite")
        for m in extra.get("recent_models", []):
            _add(m)

        # Curated fallback list so the picker is useful even before the user
        # has switched models in Kilo's native UI (which populates recent/favorite).
        # These are Kilo's own OpenRouter routing — not the shared Beacon OR key.
        _FALLBACK = [
            ("kilo/kilo-auto/free",                     "Kilo Auto — free tier"),
            ("openrouter/anthropic/claude-sonnet-4-6",  "via OpenRouter"),
            ("openrouter/anthropic/claude-opus-4-8",    "via OpenRouter"),
            ("openrouter/openai/gpt-5",                 "via OpenRouter"),
            ("openrouter/openai/gpt-4o",                "via OpenRouter"),
            ("openrouter/google/gemini-2.5-pro",        "via OpenRouter"),
            ("openrouter/google/gemini-2.5-flash",      "via OpenRouter"),
            ("openrouter/x-ai/grok-4",                  "via OpenRouter"),
            ("openrouter/openai/gpt-oss-20b",           "via OpenRouter"),
        ]
        for model_str, detail in _FALLBACK:
            _add(model_str, detail)

        return models

    # ── Lifecycle ─────────────────────────────────────────────────────────────

    async def start(self, session_id: str) -> AgentInfoBlock:
        log.info("KiloCodeAdapter.start: binary=%s cwd=%s base_url=%s resume=%s",
                 self._binary, self._cwd, self._base_url, self._resume_kilo_session_id)
        # Stamped into the kilo-serve spawn env so the server's beacon MCP child inherits it
        # (kilo, like opencode, passes the spawn env through to local MCP servers — routes the
        # sentinel call back to THIS session). One server per session, so it's fixed for life.
        self._beacon_session_id = session_id
        if not self._base_url:
            await self._spawn_server()

        if self._resume_kilo_session_id:
            self._kilo_session_id = self._resume_kilo_session_id
            if not self._cwd:
                self._cwd = await self._fetch_session_directory(self._kilo_session_id)
            log.info("KiloCodeAdapter.start: resuming kilo_session_id=%s cwd=%s",
                     self._kilo_session_id, self._cwd)
        else:
            session_data = await self._create_session()
            self._kilo_session_id = session_data.get("id", session_id)
            log.info("KiloCodeAdapter.start: kilo_session_id=%s", self._kilo_session_id)

        self._model = await self._fetch_model()
        self._context_window = context_window_for_model(self._model)
        log.info("KiloCodeAdapter.start: model=%s context_window=%d", self._model, self._context_window)

        return self.make_agent_info(
            name="Kilo Code",
            working_dirs=[self._cwd] if self._cwd else [],
        )

    async def _spawn_server(self) -> None:
        self._password = secrets.token_hex(16)
        cmd = [self._binary, "serve", "--port", "0"]
        # Sentinel: keep ~/.terminull/sentinel_mcp.py current and pass TERMINULL_SENTINEL_PORT/
        # TERMINULL_SESSION_ID into the server env so the beacon MCP child it spawns routes back.
        if type(self).cap_sentinel:
            from ..sentinel_tools import write_global_script
            write_global_script()
        env = agent_env({"KILO_SERVER_PASSWORD": self._password},
                        session_id=getattr(self, "_beacon_session_id", ""))
        log.info("KiloCodeAdapter._spawn_server: cmd=%s", " ".join(cmd))

        self._proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            env=env,
        )
        log.info("KiloCodeAdapter._spawn_server: pid=%s", self._proc.pid)

        # Parse assigned port from stdout: "...http://127.0.0.1:NNNN..."
        port: int | None = None
        for _ in range(30):
            raw = await asyncio.wait_for(self._proc.stdout.readline(), timeout=15)
            line = raw.decode().strip()
            log.debug("KiloCodeAdapter._spawn_server stdout: %s", line)
            m = re.search(r"(?:127\.0\.0\.1|localhost):(\d+)", line)
            if m:
                port = int(m.group(1))
                break

        if not port:
            raise RuntimeError("kilo serve did not print a port within timeout")

        self._base_url = f"http://127.0.0.1:{port}"
        log.info("KiloCodeAdapter._spawn_server: base_url=%s", self._base_url)

    async def _create_session(self) -> dict:
        log.info("KiloCodeAdapter._create_session: POST %s/session", self._base_url)
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{self._base_url}/session",
                json={},
                headers=self._headers(),
                timeout=10,
            )
            log.info("KiloCodeAdapter._create_session: status=%s body=%s", resp.status_code, resp.text[:200])
            resp.raise_for_status()
            return resp.json()

    async def _fetch_session_directory(self, kilo_session_id: str) -> str:
        """Return the working directory of an existing Kilo session, or '' on error."""
        try:
            async with httpx.AsyncClient() as client:
                resp = await client.get(
                    f"{self._base_url}/session/{kilo_session_id}",
                    headers=self._headers(),
                    timeout=5,
                )
                if resp.status_code == 200:
                    return resp.json().get("directory") or ""
        except Exception as e:
            log.warning("KiloCodeAdapter._fetch_session_directory: %s", e)
        return ""

    async def stop(self) -> None:
        self._stopping = True
        if self._kilo_session_id:
            try:
                async with httpx.AsyncClient() as client:
                    await client.post(
                        f"{self._base_url}/session/{self._kilo_session_id}/abort",
                        headers=self._headers(),
                        timeout=5,
                    )
            except Exception:
                pass
        if self._proc and self._proc.returncode is None:
            self._proc.terminate()
            try:
                await asyncio.wait_for(self._proc.wait(), timeout=5)
            except asyncio.TimeoutError:
                self._proc.kill()

    def clear_history(self) -> None:
        # Drop the current/resumed kilo session id; the next turn lazily creates a fresh
        # server session (see _run_turn), so context starts empty. (Was a no-op before.)
        self._resume_kilo_session_id = None
        self._kilo_session_id = ""

    # ── I/O ───────────────────────────────────────────────────────────────────

    async def send(self, text: str, attachments: list[Attachment] | None = None) -> None:
        log.info("KiloCodeAdapter.send: %d chars", len(text or ""))
        # Reject any pending questions/network prompts so the current turn can
        # unblock and complete before the new message is processed.
        for qid in list(self._pending_questions):
            await self._reject_question(qid)
        for rid in list(self._pending_network):
            await self._reject_network(rid)
        # TODO: native image attachments via {"type": "file", "uri": "data:...", "mime": ...}.
        # Until then, inline attachments as text (Mac-OCR for images + path hints) so an
        # attached photo still reaches the model rather than being dropped. See attachments.py.
        first = not self._sentinel_preamble_sent
        self._sentinel_preamble_sent = True
        _inline = self.sentinel_client_preamble(is_first_turn=first) + \
            await asyncio.to_thread(build_text_with_inline, text, attachments or [])
        parts: list[dict] = [{"type": "text", "text": _inline}]
        await self._send_queue.put({"parts": parts})

    async def respond_permission(self, permission_id: str, approved: bool) -> None:
        reply = "once" if approved else "reject"
        try:
            async with httpx.AsyncClient() as client:
                await client.post(
                    f"{self._base_url}/permission/{permission_id}/reply",
                    json={"reply": reply},
                    headers=self._headers(),
                    timeout=5,
                )
        except Exception:
            pass

    async def respond_ask_question(self, question_id: str, answers: dict) -> None:
        if question_id in self._pending_network:
            self._pending_network.discard(question_id)
            answer_val = next(iter(answers.values()), "Stop") if answers else "Stop"
            if isinstance(answer_val, list):
                answer_val = answer_val[0] if answer_val else "Stop"
            if str(answer_val).lower() == "retry":
                await self._reply_network(question_id)
            else:
                await self._reject_network(question_id)
            return

        questions = self._pending_questions.pop(question_id, [])
        # Convert {header: value} dict (iOS format) to [[label, ...], ...] (Kilo API format),
        # preserving question order so each answer aligns with its question index.
        answers_list: list[list[str]] = []
        for q in questions or [{}]:
            header = q.get("header", "")
            val = answers.get(header)
            if val is None:
                val = next(iter(answers.values()), "") if answers else ""
            answers_list.append([str(v) for v in val] if isinstance(val, list) else [str(val)])
        try:
            async with httpx.AsyncClient() as client:
                await client.post(
                    f"{self._base_url}/question/{question_id}/reply",
                    json={"answers": answers_list},
                    headers=self._headers(),
                    timeout=5,
                )
        except Exception as e:
            log.error("KiloCodeAdapter.respond_ask_question: %s", e)

    async def _reject_question(self, question_id: str) -> None:
        self._pending_questions.pop(question_id, None)
        try:
            async with httpx.AsyncClient() as client:
                await client.post(
                    f"{self._base_url}/question/{question_id}/reject",
                    headers=self._headers(),
                    timeout=5,
                )
        except Exception as e:
            log.error("KiloCodeAdapter._reject_question: %s", e)

    async def _reply_network(self, request_id: str) -> None:
        try:
            async with httpx.AsyncClient() as client:
                await client.post(
                    f"{self._base_url}/network/{request_id}/reply",
                    headers=self._headers(),
                    timeout=5,
                )
        except Exception as e:
            log.error("KiloCodeAdapter._reply_network: %s", e)

    async def _reject_network(self, request_id: str) -> None:
        self._pending_network.discard(request_id)
        try:
            async with httpx.AsyncClient() as client:
                await client.post(
                    f"{self._base_url}/network/{request_id}/reject",
                    headers=self._headers(),
                    timeout=5,
                )
        except Exception as e:
            log.error("KiloCodeAdapter._reject_network: %s", e)

    @property
    def _context_tokens(self) -> int:
        """Real context occupancy of the last prompt = fresh input + cached portion.
        This is what "% full" / auto-compaction must use (fresh `input` alone understates)."""
        return self._turn_input_tokens + self._turn_cache_tokens

    async def get_usage(self) -> dict:
        return {
            "model":                self._model,
            "input_tokens":         self._total_input_tokens,
            "output_tokens":        self._total_output_tokens,
            "cost_usd":             self._total_cost_usd,
            "turns":                self._total_turns,
            "last_input_tokens":    self._context_tokens,
            "cache_read_tokens":    self._turn_cache_tokens,
            "cache_creation_tokens": 0,
            "web_searches":         0,
            "web_fetches":          0,
            "permission_denials":   0,
        }

    async def get_status_hint(self) -> SessionStatusHintBlock | None:
        if not self._model:
            return None
        return SessionStatusHintBlock(
            model=self._model,
            turns=self._total_turns,
            cost_usd=self._total_cost_usd,
            output_tokens=self._total_output_tokens,
            cache_creation_tokens=0,
            last_input_tokens=self._context_tokens,
            context_window=self._context_window,
        )

    # ── Auto-compaction ───────────────────────────────────────────────────────

    def _should_compact(self) -> bool:
        if self._compact_threshold <= 0 or self._context_window <= 0:
            return False
        return self._context_tokens / self._context_window >= self._compact_threshold

    async def _run_compact_turn(self) -> AsyncIterator[Block]:
        """Trigger Kilo's built-in compaction agent; yield only status + SessionEndBlock."""
        pre_tokens = self._context_tokens
        self._turn_input_tokens = 0
        self._turn_cache_tokens = 0
        self._turn_output_tokens = 0
        self._turn_cost_usd = 0.0

        log.info("KiloCodeAdapter._run_compact_turn: pre=%d ctx=%d", pre_tokens, self._context_window)
        event_iter = self._sse_events()

        try:
            async with httpx.AsyncClient() as client:
                resp = await client.post(
                    f"{self._base_url}/session/{self._kilo_session_id}/prompt_async",
                    json={"mode": "compaction", "parts": [{"type": "text", "text": _COMPACT_PROMPT}]},
                    headers=self._headers(),
                    timeout=10,
                )
                if resp.status_code not in (200, 201, 202, 204):
                    log.warning("KiloCodeAdapter._run_compact_turn: HTTP %s", resp.status_code)
                    raise RuntimeError(f"compaction HTTP {resp.status_code}")
        except Exception as e:
            log.error("KiloCodeAdapter._run_compact_turn: failed to start: %s", e)
            self._turn_input_tokens = pre_tokens
            yield SystemStatusBlock(event="compaction_done")
            yield SessionEndBlock(reason="completed", input_tokens=pre_tokens,
                                  output_tokens=0, cost_usd=0.0)
            return

        kilo_sid = self._kilo_session_id
        try:
            async for etype, props in event_iter:
                event_session = props.get("sessionID") or props.get("part", {}).get("sessionID", "")
                if event_session and event_session != kilo_sid:
                    continue

                if etype == "message.updated":
                    info = props.get("info", {})
                    if info.get("role") == "assistant" and info.get("finish"):
                        tokens = info.get("tokens", {})
                        self._turn_input_tokens = tokens.get("input", 0)
                        cache = tokens.get("cache") or {}
                        self._turn_cache_tokens = cache.get("read", 0) + cache.get("write", 0)

                elif etype == "session.idle" or (
                    etype == "session.status"
                    and props.get("status", {}).get("type") == "idle"
                ):
                    post = self._context_tokens or pre_tokens
                    log.info("KiloCodeAdapter._run_compact_turn: done %d→%d tokens", pre_tokens, post)
                    yield SystemStatusBlock(event="compaction_done")
                    yield SessionEndBlock(reason="completed", input_tokens=post,
                                         output_tokens=0, cost_usd=0.0)
                    return
                # text/tool/permission events are silently consumed
        except Exception as e:
            log.error("KiloCodeAdapter._run_compact_turn: SSE error: %s", e)
            self._turn_input_tokens = pre_tokens
            yield SystemStatusBlock(event="compaction_done")
            yield SessionEndBlock(reason="completed", input_tokens=pre_tokens,
                                  output_tokens=0, cost_usd=0.0)

    def set_model(self, model: str) -> None:
        self._model = model
        self._context_window = context_window_for_model(model)
        log.info("KiloCodeAdapter: model switch to %r context_window=%d", model, self._context_window)
        if self._base_url:
            try:
                asyncio.get_event_loop().create_task(self._patch_model(model))
            except RuntimeError:
                pass

    async def _patch_model(self, model: str) -> None:
        try:
            async with httpx.AsyncClient() as client:
                resp = await client.patch(
                    f"{self._base_url}/config",
                    json={"model": model},
                    headers=self._headers(),
                    timeout=5,
                )
                log.info("KiloCodeAdapter: PATCH /config model=%r status=%s", model, resp.status_code)
        except Exception as e:
            log.warning("KiloCodeAdapter: _patch_model failed: %s", e)

    async def _fetch_model(self) -> str:
        try:
            async with httpx.AsyncClient() as client:
                resp = await client.get(
                    f"{self._base_url}/config",
                    headers=self._headers(),
                    timeout=5,
                )
                if resp.status_code == 200:
                    return resp.json().get("model") or ""
        except Exception as e:
            log.warning("KiloCodeAdapter._fetch_model: %s", e)
        return ""

    async def stream(self) -> AsyncIterator[Block]:
        while True:
            prompt_body = await self._send_queue.get()
            async for block in self._run_turn(prompt_body):
                yield block

    # ── Turn loop ─────────────────────────────────────────────────────────────

    async def _run_turn(self, prompt_body: dict) -> AsyncIterator[Block]:
        # Lazily (re)create a server session if we don't have one — e.g. after
        # clear_history() dropped it. Keeps /clear working without a full restart.
        if not self._kilo_session_id:
            data = await self._create_session()
            self._kilo_session_id = data.get("id", "")
            log.info("KiloCodeAdapter._run_turn: created fresh session=%s", self._kilo_session_id)
        log.info("KiloCodeAdapter._run_turn: session=%s", self._kilo_session_id)
        yield AudioHintBlock(state="agent_speaking")

        # Open SSE connection BEFORE sending prompt to avoid missing early events.
        event_iter = self._sse_events()

        try:
            async with httpx.AsyncClient() as client:
                log.info("KiloCodeAdapter._run_turn: POST prompt_async")
                resp = await client.post(
                    f"{self._base_url}/session/{self._kilo_session_id}/prompt_async",
                    json=prompt_body,
                    headers=self._headers(),
                    timeout=10,
                )
                log.info("KiloCodeAdapter._run_turn: prompt_async status=%s body=%s", resp.status_code, resp.text[:200])
                if resp.status_code not in (200, 201, 202, 204):
                    yield ErrorBlock(
                        message=f"Kilo /prompt_async HTTP {resp.status_code}",
                        recoverable=True,
                    )
                    yield SessionEndBlock(reason="error")
                    return
        except Exception as e:
            log.error("KiloCodeAdapter._run_turn: prompt_async exception: %s", e)
            yield ErrorBlock(message=f"Kilo connection failed: {e}", recoverable=True)
            yield SessionEndBlock(reason="error")
            return

        # Reset per-turn state
        self._stopping = False
        self._part_types.clear()
        self._text_parts.clear()
        self._thinking_parts.clear()
        self._pending_tools.clear()
        self._pending_questions.clear()
        self._pending_network.clear()
        self._turn_input_tokens = 0
        self._turn_output_tokens = 0
        self._turn_cost_usd = 0.0

        kilo_sid = self._kilo_session_id
        try:
            async for etype, props in event_iter:
                # Filter to our session (skip events that belong to other sessions)
                event_session = props.get("sessionID") or props.get("part", {}).get("sessionID", "")
                if event_session and event_session != kilo_sid:
                    log.info("KiloCodeAdapter: skip event %s from session %s (ours=%s)", etype, event_session, kilo_sid)
                    continue

                if etype in ("message.part.updated", "message.part.updated_v1"):
                    log.debug("KiloCodeAdapter: ← %s %s", etype, str(props)[:2000])
                else:
                    log.debug("KiloCodeAdapter: ← %s %s", etype, str(props)[:200])
                done = False
                async for block in self._parse_event(etype, props):
                    if isinstance(block, SessionEndBlock):
                        done = True
                    yield block

                if done:
                    log.info("KiloCodeAdapter._run_turn: turn complete")
                    break
        except Exception as e:
            log.error("KiloCodeAdapter._run_turn: SSE exception: %s", e, exc_info=True)
            yield ErrorBlock(message=str(e), recoverable=True)
            yield SessionEndBlock(reason="error")

    async def _sse_events(self) -> AsyncIterator[tuple[str, dict]]:
        """Yield (event_type, properties) from GET /event until turn completes."""
        log.info("KiloCodeAdapter._sse_events: connecting to %s/event", self._base_url)
        async with httpx.AsyncClient(
            timeout=httpx.Timeout(300, connect=10)
        ) as client:
            async with client.stream(
                "GET",
                f"{self._base_url}/event",
                headers={**self._headers(), "Accept": "text/event-stream"},
            ) as response:
                log.info("KiloCodeAdapter._sse_events: SSE status=%s", response.status_code)
                try:
                    async for line in response.aiter_lines():
                        if not line.startswith("data: "):
                            continue
                        raw = line[6:].strip()
                        if not raw:
                            continue
                        try:
                            event = json.loads(raw)
                        except Exception:
                            log.warning("KiloCodeAdapter._sse_events: bad JSON: %s", raw[:200])
                            continue

                        etype = event.get("type", "")
                        # SyncEvents on the bus wrap payload in {type, properties:{...}}
                        # BusEvents use {type, properties:{...}} directly.
                        props = event.get("properties", {})
                        yield etype, props
                except GeneratorExit:
                    log.info("KiloCodeAdapter._sse_events: generator closed")
                    raise
                except Exception as e:
                    log.error("KiloCodeAdapter._sse_events: stream error: %s", e)
                    raise

    # ── Tool block helpers ────────────────────────────────────────────────────

    async def _emit_tool_running(self, tool_name: str, call_id: str, args: dict) -> AsyncIterator[Block]:
        block_id = self._pending_tools.get(call_id) or str(uuid.uuid4())
        self._pending_tools[call_id] = block_id
        safe_args = args if isinstance(args, dict) else {}
        lbl, subj = tool_label_subject(tool_name, safe_args)
        log.debug("KiloCodeAdapter: → ToolCallBlock(running) tool=%s block_id=%s", tool_name, block_id)
        yield AudioHintBlock(state="agent_silent")
        yield ToolCallBlock(id=block_id, name=tool_name, args=safe_args,
                            status="running", label=lbl, subject=subj)

    async def _emit_tool_done(self, tool_name: str, call_id: str, args: dict,
                              result_text: str, is_error: bool) -> AsyncIterator[Block]:
        block_id = self._pending_tools.pop(call_id, str(uuid.uuid4()))
        safe_args = args if isinstance(args, dict) else {}
        lbl, subj = tool_label_subject(tool_name, safe_args)
        log.debug("KiloCodeAdapter: → ToolCallBlock(done) tool=%s is_error=%s", tool_name, is_error)
        yield ToolCallBlock(id=block_id, name=tool_name, args=safe_args,
                            status="error" if is_error else "done", label=lbl, subject=subj)
        yield tool_result_or_large(tool_call_id=call_id, content=result_text,
                                   fetch_store=self._fetch_store, is_error=is_error)
        yield AudioHintBlock(state="agent_speaking")

    # ── Event parser ─────────────────────────────────────────────────────────

    async def _parse_event(self, etype: str, props: dict) -> AsyncIterator[Block]:
        if etype in ("server.connected", "server.heartbeat"):
            return

        # ── Text / reasoning streaming ───────────────────────────────────────
        elif etype == "message.part.delta":
            part_id = props.get("partID", "")
            field = props.get("field", "")
            delta = props.get("delta", "")

            if field != "text":
                return

            part_type = self._part_types.get(part_id, "text")

            # iOS APPENDS done=False partials (the universal delta-stream contract),
            # so yield only the NEW delta, never the accumulated buffer — the buffer
            # duplicated text per event ("The" → "TheThe quick" → …), corrected only
            # by the done=True snapshot in message.part.updated. Keep accumulating
            # internally so the final done=True block carries the full text.
            if part_type == "reasoning":
                self._thinking_parts[part_id] = self._thinking_parts.get(part_id, "") + delta
                yield ThinkingBlock(thinking=delta, done=False)
            else:
                if part_id not in self._text_parts:
                    self._text_parts[part_id] = (str(uuid.uuid4()), "")
                block_id, accumulated = self._text_parts[part_id]
                accumulated += delta
                self._text_parts[part_id] = (block_id, accumulated)
                yield TextBlock(text=delta, done=False, id=block_id)

        # ── Part state updates (tool calls, finalized text/reasoning) ────────
        elif etype in ("message.part.updated", "message.part.updated_v1"):
            part = props.get("part", {})
            part_id = part.get("id", "")
            part_type = part.get("type", "")
            log.info("KiloCodeAdapter: message.part.updated part_id=%s part_type=%s part_keys=%s",
                     part_id, part_type, list(part.keys()))

            # Record part type for future delta events
            if part_id and part_type:
                self._part_types[part_id] = part_type

            if part_type == "text":
                text = part.get("text", "")
                if part_id in self._text_parts:
                    block_id, _ = self._text_parts.pop(part_id)
                    yield TextBlock(text=text, done=True, id=block_id)
                    # Emit any structured blocks (e.g. CodeBlock) from the finalized text
                    for block in split_markdown(text):
                        if not isinstance(block, TextBlock):
                            yield block
                elif text:
                    for block in split_markdown(text):
                        yield block

            elif part_type == "reasoning":
                text = part.get("text", self._thinking_parts.pop(part_id, ""))
                if text:
                    yield ThinkingBlock(thinking=text, done=True)

            elif part_type == "tool-invocation":
                # Vercel AI SDK wire format: toolInvocation.{state, toolCallId, toolName, args, result}
                ti = part.get("toolInvocation", {})
                state = ti.get("state", "")
                call_id = ti.get("toolCallId", "")
                tool_name = ti.get("toolName", "")
                args = ti.get("args") or {}
                log.info("KiloCodeAdapter: tool-invocation part_id=%s state=%s tool=%s call_id=%s",
                         part_id, state, tool_name, call_id)
                if state in ("call", "partial-call"):
                    async for block in self._emit_tool_running(tool_name, call_id, args):
                        yield block
                elif state == "result":
                    result = ti.get("result", "")
                    is_error = isinstance(result, dict) and bool(result.get("isError"))
                    result_text = result if isinstance(result, str) else json.dumps(result)
                    async for block in self._emit_tool_done(tool_name, call_id, args, result_text, is_error):
                        yield block

            # Kilo's native tool format: {type:"tool", tool:name, callID:..., state:{status, input, output}}
            elif part_type == "tool":
                tool_name = part.get("tool", "")
                call_id = part.get("callID", "")
                state_obj = part.get("state", {})
                status = state_obj.get("status", "")
                args = state_obj.get("input") or {}
                log.info("KiloCodeAdapter: tool part_id=%s status=%s tool=%s call_id=%s",
                         part_id, status, tool_name, call_id)
                if status == "running":
                    async for block in self._emit_tool_running(tool_name, call_id, args):
                        yield block
                elif status == "completed":
                    output = state_obj.get("output", "")
                    is_error = isinstance(output, dict) and bool(output.get("isError"))
                    result_text = output if isinstance(output, str) else json.dumps(output)
                    async for block in self._emit_tool_done(tool_name, call_id, args, result_text, is_error):
                        yield block

        # ── Permission request ───────────────────────────────────────────────
        elif etype == "permission.asked":
            perm_id = props.get("id", "")
            permission = props.get("permission", "")
            patterns = props.get("patterns", [])
            metadata = props.get("metadata", {})
            details_parts = list(patterns)
            if metadata:
                details_parts.append(json.dumps(metadata, separators=(",", ":")))
            yield PermissionRequestBlock(
                id=perm_id,
                action=permission,
                risk="medium",
                details=", ".join(details_parts),
            )

        # ── Question (ask_followup_question tool) ────────────────────────────
        elif etype == "question.asked":
            request_id = props.get("id", "")
            questions = props.get("questions", [])
            self._pending_questions[request_id] = questions
            if questions:
                q = questions[0]
                opts = [o["label"] for o in q.get("options", []) if "label" in o]
                yield InputRequestBlock(
                    id=request_id,
                    prompt=q.get("question", ""),
                    input_type="choice" if opts else "text",
                    options=opts,
                )
                yield AudioHintBlock(state="agent_silent")

        elif etype == "question.rejected":
            self._pending_questions.pop(props.get("requestID", ""), None)

        # ── Suggestions (non-blocking follow-up actions) ─────────────────────
        elif etype == "suggestion.shown":
            actions = props.get("actions", [])
            steps = [a.get("prompt") or a.get("label", "") for a in actions if a.get("prompt") or a.get("label")]
            if steps:
                yield SuggestionsBlock(steps=steps, context=props.get("text", ""))

        # ── Network recovery (connection lost mid-turn) ───────────────────────
        elif etype == "session.network.asked":
            request_id = props.get("id", "")
            self._pending_network.add(request_id)
            yield InputRequestBlock(
                id=request_id,
                prompt=props.get("message", "Network connection lost."),
                input_type="choice",
                options=["Retry", "Stop"],
            )
            yield AudioHintBlock(state="agent_silent")

        elif etype in ("session.network.restored", "session.network.replied", "session.network.rejected"):
            self._pending_network.discard(props.get("requestID", "") or props.get("id", ""))

        # ── Config change (model switched in Kilo UI) ─────────────────────────
        elif etype == "global.config.updated":
            self._model = await self._fetch_model()
            self._context_window = context_window_for_model(self._model)

        # ── Per-turn usage (message.updated for final assistant message) ────────
        elif etype == "message.updated":
            info = props.get("info", {})
            if info.get("role") == "assistant" and info.get("finish"):
                tokens = info.get("tokens", {})
                self._turn_input_tokens = tokens.get("input", 0)
                cache = tokens.get("cache") or {}
                self._turn_cache_tokens = cache.get("read", 0) + cache.get("write", 0)
                self._turn_output_tokens = tokens.get("output", 0)
                self._turn_cost_usd = float(info.get("cost", 0.0))

        # ── Turn completion ──────────────────────────────────────────────────
        elif etype == "session.idle":
            self._total_input_tokens += self._turn_input_tokens
            self._total_output_tokens += self._turn_output_tokens
            self._total_cost_usd += self._turn_cost_usd
            self._total_turns += 1
            yield AudioHintBlock(state="agent_silent")
            if self._should_compact():
                yield SystemStatusBlock(event="compacting")
                async for block in self._run_compact_turn():
                    yield block
            else:
                yield SessionEndBlock(
                    reason="completed",
                    input_tokens=self._turn_input_tokens,
                    output_tokens=self._turn_output_tokens,
                    cost_usd=self._turn_cost_usd,
                )

        elif etype == "session.status":
            status = props.get("status", {})
            status_type = status.get("type")
            if status_type == "idle":
                self._total_input_tokens += self._turn_input_tokens
                self._total_output_tokens += self._turn_output_tokens
                self._total_cost_usd += self._turn_cost_usd
                self._total_turns += 1
                yield AudioHintBlock(state="agent_silent")
                if self._should_compact():
                    yield SystemStatusBlock(event="compacting")
                    async for block in self._run_compact_turn():
                        yield block
                else:
                    yield SessionEndBlock(
                        reason="completed",
                        input_tokens=self._turn_input_tokens,
                        output_tokens=self._turn_output_tokens,
                        cost_usd=self._turn_cost_usd,
                    )
            elif status_type == "retry":
                attempt = status.get("attempt", 1)
                msg = status.get("message", "Retrying…")
                yield ThinkingBlock(thinking=f"Retry {attempt}: {msg}", done=True)

        # ── Error ────────────────────────────────────────────────────────────
        elif etype == "session.error":
            if self._stopping:
                yield AudioHintBlock(state="agent_silent")
                yield SessionEndBlock(reason="user_stopped")
                return
            error = props.get("error", {})
            msg = (
                error.get("message")
                or error.get("data", {}).get("message")
                or str(error)
            )
            yield ErrorBlock(message=msg, recoverable=True)
            yield AudioHintBlock(state="agent_silent")
            yield SessionEndBlock(reason="error")

        # ── Todo list ────────────────────────────────────────────────────────
        elif etype == "todo.updated":
            todos = props.get("todos", [])
            if todos:
                yield self.todo_block(self._kilo_session_id, todos)

        else:
            log.info("KiloCodeAdapter: unhandled event %s props=%s", etype, str(props)[:200])

    # ── Helpers ───────────────────────────────────────────────────────────────

    def _headers(self) -> dict[str, str]:
        h: dict[str, str] = {}
        if self._password:
            token = base64.b64encode(f"kilo:{self._password}".encode()).decode()
            h["Authorization"] = f"Basic {token}"
        if self._cwd:
            h["x-kilo-directory"] = self._cwd
        return h
