From 77f5f34be59bbb43008891f808cce0714315cb20 Mon Sep 17 00:00:00 2001
From: Diego Jaramillo Celis <7494599+djcelis@users.noreply.github.com>
Date: Tue, 29 Sep 2026 17:55:30 -0400
Subject: [PATCH 1/2] Add Kilo CLI as an agent provider

Kilo (@kilocode/cli) is an OpenCode fork sharing config schema, provider
model, and plugin system. This adds a kilo agent module that reuses the
shared opencode_models discovery, isolates config via KILO_CONFIG, gates
on kilo >= 7.3.1, and spawns via subprocess_cross_os.popen for Windows
npm-binary resolution.
---
 src/ucode/agents/__init__.py |  12 +-
 src/ucode/agents/kilo.py     | 469 +++++++++++++++++++++++++++
 src/ucode/cli.py             |  44 ++-
 tests/test_agent_kilo.py     | 596 +++++++++++++++++++++++++++++++++++
 tests/test_agents_init.py    |   2 +-
 tests/test_cli.py            |   6 +-
 6 files changed, 1116 insertions(+), 13 deletions(-)
 create mode 100644 src/ucode/agents/kilo.py
 create mode 100644 tests/test_agent_kilo.py

diff --git a/src/ucode/agents/__init__.py b/src/ucode/agents/__init__.py
index 3091a948f..27525ed7d 100644
--- a/src/ucode/agents/__init__.py
+++ b/src/ucode/agents/__init__.py
@@ -41,7 +41,7 @@
     status_subprocess_stdout,
 )
 
-from . import claude, codex, copilot, gemini, opencode, pi
+from . import claude, codex, copilot, gemini, kilo, opencode, pi
 from .args import LaunchOptions as LaunchOptions
 from .args import explicit_model_arg_value as explicit_model_arg_value
 
@@ -50,6 +50,7 @@
     "claude": claude,
     "gemini": gemini,
     "opencode": opencode,
+    "kilo": kilo,
     "copilot": copilot,
     "pi": pi,
 }
@@ -68,6 +69,8 @@
     "gemini": "gemini",
     "gemini-cli": "gemini",
     "opencode": "opencode",
+    "kilo": "kilo",
+    "kilocode": "kilo",
     "copilot": "copilot",
     "pi": "pi",
 }
@@ -116,7 +119,7 @@ def normalize_tool(tool: str) -> str:
     normalized = TOOL_ALIASES.get(tool.strip().lower())
     if not normalized:
         raise RuntimeError(
-            f"Unsupported tool '{tool}'. Use one of: codex, claude, gemini, opencode, copilot, pi."
+            f"Unsupported tool '{tool}'. Use one of: codex, claude, gemini, opencode, kilo, copilot, pi."
         )
     return normalized
 
@@ -444,6 +447,8 @@ def configure_tool(
             result = copilot.write_tool_config(state, model)
         elif tool == "pi":
             result = pi.write_tool_config(state, model)
+        elif tool == "kilo":
+            result = kilo.write_tool_config(state, model)
         else:
             result = opencode.write_tool_config(state, model)
     # gemini/opencode/copilot/pi return (state, token); codex/claude return state
@@ -485,7 +490,7 @@ def check_gateway_endpoint(state: dict, tool: str) -> bool:
     """V2-only: a tool is available iff we discovered models for it."""
     if tool == "claude":
         return bool(state.get("claude_models"))
-    if tool == "opencode":
+    if tool in ("opencode", "kilo"):
         return bool(state.get("opencode_models"))
     if tool == "codex":
         return bool(state.get("codex_models"))
@@ -505,6 +510,7 @@ def check_gateway_endpoint(state: dict, tool: str) -> bool:
 _TOOL_DISCOVERY_SOURCES: dict[str, tuple[str, ...]] = {
     "claude": ("claude",),
     "opencode": ("claude", "gemini", "oss"),
+    "kilo": ("claude", "gemini", "oss"),
     "codex": ("codex",),
     "gemini": ("gemini",),
     "copilot": ("claude", "codex"),
diff --git a/src/ucode/agents/kilo.py b/src/ucode/agents/kilo.py
new file mode 100644
index 000000000..4dca77d1f
--- /dev/null
+++ b/src/ucode/agents/kilo.py
@@ -0,0 +1,469 @@
+"""Kilo agent: writes kilo.json with Databricks-backed providers.
+
+Kilo CLI (@kilocode/cli) is a fork of OpenCode; it shares OpenCode's config
+schema, provider model, and plugin system. This module mirrors opencode.py,
+reading the same discovered `opencode_models` view from state, and isolating
+Kilo via a single-file KILO_CONFIG (proven exclusive on kilo 7.2.x, whereas
+KILO_CONFIG_DIR merges the user's real config dir underneath)."""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import signal
+
+from ucode.config_io import (
+    APP_DIR,
+    ToolSpec,
+    apply_json_mcp_diff,
+    backup_existing_file,
+    deep_merge_dict,
+    read_json_safe,
+    write_json_file,
+    write_text_file,
+)
+from ucode.databricks import (
+    build_auth_token_argv,
+    build_opencode_base_urls,
+    get_databricks_token,
+    model_token_limits,
+)
+from ucode.os_compatibility import subprocess_cross_os
+from ucode.state import mark_tool_managed, save_state
+from ucode.telemetry import agent_version, ug_version
+
+from .args import LaunchOptions, explicit_model_arg_value, has_explicit_model_arg
+
+KILO_CONFIG_DIR = APP_DIR / "kilo-config"
+KILO_CONFIG_INNER_DIR = KILO_CONFIG_DIR
+KILO_CONFIG_PATH = KILO_CONFIG_INNER_DIR / "kilo.json"
+KILO_BACKUP_PATH = APP_DIR / "kilo-config.backup.json"
+KILO_AUTH_PLUGIN_PATH = KILO_CONFIG_INNER_DIR / "plugin" / "ucode-auth.js"
+KILO_NPM_PACKAGE = "@kilocode/cli"
+# Minimum Kilo version whose plugin runtime invokes the `config()` hook that our
+# auth plugin (ucode-auth.js) relies on to refresh Databricks tokens.
+# Verified empirically (plugin `config()` probe):
+#   7.2.1  -> hook does NOT fire (unsupported; 7.2.x is a noisy patch series)
+#   7.2.25 -> hook fires
+#   7.3.1  -> hook fires  <-- chosen floor: earliest clean minor boundary that works
+#   7.4.1  -> hook fires
+MINIMUM_KILO_VERSION = (7, 3, 1)
+MINIMUM_KILO_VERSION_TEXT = "7.3.1"
+
+SPEC: ToolSpec = {
+    "binary": "kilo",
+    "package": f"{KILO_NPM_PACKAGE}@7",
+    "display": "Kilo",
+    "config_path": KILO_CONFIG_PATH,
+    "backup_path": KILO_BACKUP_PATH,
+}
+
+PROVIDER_KEYS: list[list[str]] = [
+    ["provider", "databricks-anthropic"],
+    ["provider", "databricks-google"],
+    ["provider", "databricks-oss"],
+]
+
+_AUTH_PLUGIN_TEMPLATE = """// Generated by ucode. Keep Databricks auth fresh for model requests.
+import { execFile } from "node:child_process"
+import { promisify } from "node:util"
+
+const DATABRICKS_PROVIDERS = new Set([
+  "databricks-anthropic",
+  "databricks-google",
+  "databricks-oss",
+])
+const AUTH_COMMAND = __AUTH_COMMAND__
+const REFRESH_SKEW_MS = 120_000
+const run = promisify(execFile)
+
+let accessToken
+let expiresAt = 0
+let refreshPromise
+
+function cacheToken(value) {
+  accessToken = value
+  expiresAt = Infinity
+  try {
+    const claims = JSON.parse(Buffer.from(value.split(".")[1], "base64url").toString())
+    if (typeof claims.exp === "number") expiresAt = claims.exp * 1000
+  } catch {}
+}
+
+async function mintToken() {
+  try {
+    const { stdout } = await run(AUTH_COMMAND[0], AUTH_COMMAND.slice(1), { encoding: "utf8" })
+    const token = stdout.trim()
+    if (!token) throw new Error("returned an empty token")
+    cacheToken(token)
+  } catch (error) {
+    const detail = String(error.stderr || error.message || "").trim()
+    throw new Error("ug auth-token failed" + (detail ? ": " + detail : ""))
+  }
+}
+
+function refreshToken() {
+  if (!refreshPromise) refreshPromise = mintToken().finally(() => { refreshPromise = undefined })
+  return refreshPromise
+}
+
+function requestWithToken(input, init, token) {
+  const headers = new Headers(input instanceof Request ? input.headers : undefined)
+  new Headers(init?.headers).forEach((value, key) => headers.set(key, value))
+  headers.set("Authorization", "Bearer " + token)
+  return { ...init, headers }
+}
+
+async function databricksFetch(input, init) {
+  if (!accessToken || expiresAt <= Date.now() + REFRESH_SKEW_MS) await refreshToken()
+  const attemptedToken = accessToken
+  const response = await fetch(input, requestWithToken(input, init, attemptedToken))
+  if (response.status !== 401) return response
+
+  // Another concurrent request may already have refreshed this token.
+  if (accessToken === attemptedToken) await refreshToken()
+  return fetch(input, requestWithToken(input, init, accessToken))
+}
+
+export const UcodeDatabricksAuth = async () => ({
+  config: async (config) => {
+    for (const providerID of DATABRICKS_PROVIDERS) {
+      const options = config.provider?.[providerID]?.options
+      if (!options) continue
+      if (!accessToken && typeof options.apiKey === "string") cacheToken(options.apiKey)
+      options.fetch = databricksFetch
+    }
+  },
+})
+"""
+
+
+def _parse_version(value: str) -> tuple[int, int, int] | None:
+    match = re.search(r"(\d+)\.(\d+)\.(\d+)", value)
+    if match is None:
+        return None
+    major, minor, patch = match.groups()
+    return int(major), int(minor), int(patch)
+
+
+def _minimum_version_message() -> str | None:
+    installed = agent_version(SPEC["binary"])
+    parsed = _parse_version(installed)
+    if parsed is None or parsed >= MINIMUM_KILO_VERSION:
+        return None
+    return (
+        f"Kilo {installed} is too old. ucode requires Kilo "
+        f"{MINIMUM_KILO_VERSION_TEXT} or newer for renewable Databricks authentication."
+    )
+
+
+def minimum_version_error() -> str | None:
+    message = _minimum_version_message()
+    if message is None:
+        return None
+    return f"{message} Update it with `npm install -g {SPEC['package']}`."
+
+
+def render_auth_plugin(state: dict) -> str:
+    """Render the local Kilo plugin that refreshes Databricks auth on demand."""
+    argv = build_auth_token_argv(
+        state["workspace"],
+        state.get("profile"),
+        use_pat=bool(state.get("use_pat")),
+    )
+    # A 401 must not return the same still-unexpired cached credential.
+    argv.append("--force-refresh")
+    return _AUTH_PLUGIN_TEMPLATE.replace("__AUTH_COMMAND__", json.dumps(argv))
+
+
+def write_auth_plugin(state: dict) -> None:
+    """Install the auto-discovered hook in ucode's isolated Kilo config."""
+    # Derive the path so tests that redirect KILO_CONFIG_PATH stay isolated.
+    path = KILO_CONFIG_PATH.parent / "plugin" / KILO_AUTH_PLUGIN_PATH.name
+    write_text_file(path, render_auth_plugin(state))
+
+
+def _resolve_model_selector(model: str, opencode_models: dict[str, list[str]]) -> str:
+    """Return a Kilo model selector in provider/model form when possible."""
+    if model.startswith(("databricks-anthropic/", "databricks-google/", "databricks-oss/")):
+        return model
+
+    anthropic_models = opencode_models.get("anthropic") or []
+    if model in anthropic_models:
+        return f"databricks-anthropic/{model}"
+
+    gemini_models = opencode_models.get("gemini") or []
+    if model in gemini_models:
+        return f"databricks-google/{model}"
+
+    oss_models = opencode_models.get("oss") or []
+    if model in oss_models:
+        return f"databricks-oss/{model}"
+
+    return model
+
+
+def resolve_explicit_model(model: str, state: dict) -> str:
+    """Resolve and validate a model explicitly selected for a Kilo launch.
+
+    Databricks model ids are exposed in family buckets in ucode state, while
+    Kilo addresses those models through managed provider ids. Other
+    provider/model selectors belong to Kilo and are passed through.
+    """
+    selector = model.strip()
+    if not selector:
+        raise RuntimeError("Kilo model must not be empty.")
+
+    opencode_models = state.get("opencode_models") or {}
+    resolved = _resolve_model_selector(selector, opencode_models)
+    if "/" not in selector:
+        if resolved != selector:
+            return resolved
+        raise RuntimeError(
+            f"Kilo model '{selector}' is not configured. "
+            "Choose a discovered model id or pass a provider/model selector."
+        )
+
+    provider, _, model_id = selector.partition("/")
+    if not provider or not model_id:
+        raise RuntimeError("Kilo model selector must use provider/model form.")
+    managed_providers = {path[-1] for path in PROVIDER_KEYS}
+    if provider not in managed_providers:
+        return selector
+
+    if _resolve_model_selector(model_id, opencode_models) == selector:
+        return selector
+    raise RuntimeError(
+        f"Kilo model '{selector}' is not configured for managed provider '{provider}'."
+    )
+
+
+def _oss_model_overlay(model: str, ua_header: dict[str, str]) -> dict:
+    """Per-model overlay for an OSS model entry.
+
+    All OSS models carry the User-Agent header; models with known token limits
+    also pin `limit` (context + output) so Kilo clamps `max_tokens` to a
+    value the gateway accepts. Kilo's schema requires both fields together,
+    so the limits table always supplies both."""
+    overlay: dict = {"headers": ua_header}
+    limits = model_token_limits(model)
+    if limits is not None:
+        overlay["limit"] = limits
+    return overlay
+
+
+def render_overlay(
+    model: str,
+    token: str,
+    opencode_base_urls: dict[str, str],
+    opencode_models: dict[str, list[str]],
+) -> tuple[dict, list[list[str]]]:
+    """Return (overlay, managed_key_paths) for kilo.json."""
+    auth_headers = {"Authorization": f"Bearer {token}"}
+    # Kilo hardcodes `User-Agent: opencode/<ver>` in session/llm.ts for
+    # every provider, after the AI SDK's combineHeaders. The provider-level
+    # `headers` are clobbered by that injection, but per-model `headers` are
+    # merged AFTER and win — so the UA must live on each model entry.
+    ua_header = {
+        "User-Agent": f"ucode/{ug_version()} kilo/{agent_version('kilo')}",
+    }
+
+    anthropic_models = opencode_models.get("anthropic") or []
+    gemini_models = opencode_models.get("gemini") or []
+    oss_models = opencode_models.get("oss") or []
+
+    providers: dict = {}
+    keys: list[list[str]] = [["model"]]
+    if anthropic_models:
+        # @ai-sdk/anthropic injects `eager_input_streaming: true` on tool defs;
+        # the Databricks gateway's strict validator rejects it. opencode's
+        # auto-disable in transform.ts skips models whose id contains "claude",
+        # so we opt out per-model. The setting lives in per-call providerOptions,
+        # which opencode reads from `models.<m>.options`, not provider `options`.
+        anthropic_model_overlay = {
+            "headers": ua_header,
+            "options": {"toolStreaming": False},
+        }
+        providers["databricks-anthropic"] = {
+            "npm": "@ai-sdk/anthropic",
+            "options": {
+                "baseURL": opencode_base_urls["anthropic"],
+                "apiKey": token,
+                "headers": auth_headers,
+            },
+            "models": dict.fromkeys(anthropic_models, anthropic_model_overlay),
+        }
+        keys.append(["provider", "databricks-anthropic"])
+    if gemini_models:
+        providers["databricks-google"] = {
+            "npm": "@ai-sdk/google",
+            "options": {
+                "baseURL": opencode_base_urls["gemini"],
+                "apiKey": token,
+                "headers": auth_headers,
+            },
+            "models": {m: {"headers": ua_header} for m in gemini_models},
+        }
+        keys.append(["provider", "databricks-google"])
+    if oss_models:
+        providers["databricks-oss"] = {
+            "npm": "@ai-sdk/openai",
+            "options": {
+                "baseURL": opencode_base_urls["oss"],
+                "apiKey": token,
+                "headers": auth_headers,
+            },
+            "models": {m: _oss_model_overlay(m, ua_header) for m in oss_models},
+        }
+        keys.append(["provider", "databricks-oss"])
+
+    overlay: dict = {"model": _resolve_model_selector(model, opencode_models)}
+    if providers:
+        overlay["provider"] = providers
+    return overlay, keys
+
+
+def write_tool_config(
+    state: dict,
+    model: str,
+    token: str | None = None,
+) -> tuple[dict, str]:
+    backup_existing_file(KILO_CONFIG_PATH, KILO_BACKUP_PATH)
+    if token is None:
+        token = get_databricks_token(state["workspace"], state.get("profile"))
+    opencode_base_urls = state.get("base_urls", {}).get("opencode") or build_opencode_base_urls(
+        state["workspace"]
+    )
+    overlay, managed_keys = render_overlay(
+        model,
+        token,
+        opencode_base_urls,
+        state.get("opencode_models") or {},
+    )
+    existing = read_json_safe(KILO_CONFIG_PATH)
+    write_auth_plugin(state)
+    providers = existing.get("provider")
+    if isinstance(providers, dict):
+        for stale in (
+            "databricks-anthropic",
+            "databricks-google",
+            "databricks-openai",
+            "databricks-oss",
+        ):
+            providers.pop(stale, None)
+    merged = deep_merge_dict(existing, overlay)
+    write_json_file(KILO_CONFIG_PATH, merged)
+    state = mark_tool_managed(state, "kilo", managed_keys)
+    save_state(state)
+    return state, token
+
+
+def build_mcp_server_entry(argv: list[str]) -> dict:
+    # A `local` MCP server runs a command over stdio; `command` is the full
+    # argv. ug registers the `ug mcp-proxy ...` bridge here so Kilo
+    # never speaks HTTP+bearer directly — the proxy mints fresh tokens itself.
+    return {
+        "type": "local",
+        "command": list(argv),
+        "enabled": True,
+    }
+
+
+def write_mcp_server_config(name: str, argv: list[str]) -> bool:
+    backup_existing_file(KILO_CONFIG_PATH, KILO_BACKUP_PATH)
+    existing = read_json_safe(KILO_CONFIG_PATH)
+    mcp_servers = existing.get("mcp")
+    if not isinstance(mcp_servers, dict):
+        mcp_servers = {}
+    removed = name in mcp_servers
+    mcp_servers[name] = build_mcp_server_entry(argv)
+    existing["mcp"] = mcp_servers
+    write_json_file(KILO_CONFIG_PATH, existing)
+    return removed
+
+
+def remove_mcp_server_config(name: str) -> bool:
+    existing = read_json_safe(KILO_CONFIG_PATH)
+    mcp_servers = existing.get("mcp")
+    if not isinstance(mcp_servers, dict) or name not in mcp_servers:
+        return False
+    mcp_servers.pop(name)
+    existing["mcp"] = mcp_servers
+    write_json_file(KILO_CONFIG_PATH, existing)
+    return True
+
+
+def write_user_mcp_servers(add: dict[str, dict], remove: set[str]) -> set[str]:
+    """Apply ``add``/``remove`` to Kilo's `mcp` table in a single read-modify-write. Returns the names actually removed."""
+    return apply_json_mcp_diff(KILO_CONFIG_PATH, "mcp", add, remove, backup_path=KILO_BACKUP_PATH)
+
+
+def default_model(state: dict) -> str | None:
+    if isinstance(state.get("opencode_default_model"), str):
+        return state.get("opencode_default_model")
+    opencode_models = state.get("opencode_models") or {}
+    anthropic = opencode_models.get("anthropic") or []
+    if anthropic:
+        return anthropic[0]
+    gemini = opencode_models.get("gemini") or []
+    if gemini:
+        return gemini[0]
+    oss = opencode_models.get("oss") or []
+    return oss[0] if oss else None
+
+
+def _configure_launch(state: dict, model: str | None = None) -> str:
+    model = model or default_model(state)
+    if not model:
+        raise RuntimeError("No Kilo model is configured.")
+    _, token = write_tool_config(state, model)
+    return token
+
+
+def build_runtime_env(token: str, state: dict | None = None) -> dict[str, str]:
+    env = os.environ.copy()
+    env["OAUTH_TOKEN"] = token
+    # Isolation: unlike OpenCode (which neutralizes XDG_CONFIG_HOME), Kilo pins a
+    # single config file via KILO_CONFIG. Verified empirically against kilo 7.2.25:
+    # with KILO_CONFIG=<file> set, Kilo loads ONLY that file and does NOT merge the
+    # user's real ~/.config/kilo. (KILO_CONFIG_DIR alone is additive/merges the real
+    # dir, so it is used only to point Kilo at our plugin directory, not for config
+    # isolation.) This keeps a co-installed OpenCode's config out of Kilo's merge chain.
+    env["KILO_CONFIG"] = str(KILO_CONFIG_PATH)
+    env["KILO_CONFIG_DIR"] = str(KILO_CONFIG_INNER_DIR)
+    return env
+
+
+def launch(state: dict, tool_args: list[str], *, options: LaunchOptions) -> None:
+    """Launch Kilo with the selected model and on-demand token refresh."""
+    model = explicit_model_arg_value(tool_args) or options.user_pinned_model
+    if model is not None:
+        model = resolve_explicit_model(model, state)
+    token = _configure_launch(state, model)
+    env = build_runtime_env(token, state)
+
+    if model is not None and not has_explicit_model_arg(tool_args):
+        try:
+            separator = tool_args.index("--")
+        except ValueError:
+            separator = len(tool_args)
+        tool_args = [
+            *tool_args[:separator],
+            "--model",
+            model,
+            *tool_args[separator:],
+        ]
+
+    proc = subprocess_cross_os.popen([SPEC["binary"], *tool_args], env=env)
+    try:
+        returncode = proc.wait()
+    except KeyboardInterrupt:
+        proc.send_signal(signal.SIGINT)
+        returncode = proc.wait()
+
+    raise SystemExit(returncode)
+
+
+def validate_cmd(binary: str) -> list[str]:
+    return [binary, "run", "say hi in 5 words or less"]
diff --git a/src/ucode/cli.py b/src/ucode/cli.py
index 656abd746..3aa3baaea 100644
--- a/src/ucode/cli.py
+++ b/src/ucode/cli.py
@@ -175,10 +175,10 @@
 CustomOAuthConfig = custom_oauth.CustomOAuthConfig
 
 _DISCOVERY_CONSUMERS: dict[str, tuple[str, ...]] = {
-    "claude": ("claude", "opencode", "copilot", "pi"),
+    "claude": ("claude", "opencode", "kilo", "copilot", "pi"),
     "codex": ("codex", "copilot", "pi"),
-    "gemini": ("gemini", "opencode", "pi"),
-    "oss": ("opencode",),
+    "gemini": ("gemini", "opencode", "kilo", "pi"),
+    "oss": ("opencode", "kilo"),
 }
 
 
@@ -602,16 +602,23 @@ def configure_shared_state(
         print_warning(f"Model service: {model_service_probe.detail}")
 
     want_claude = (
-        fetch_all or "claude" in tools or "opencode" in tools or "copilot" in tools or "pi" in tools
+        fetch_all
+        or "claude" in tools
+        or "opencode" in tools
+        or "kilo" in tools
+        or "copilot" in tools
+        or "pi" in tools
+    )
+    want_gemini = (
+        fetch_all or "gemini" in tools or "opencode" in tools or "kilo" in tools or "pi" in tools
     )
-    want_gemini = fetch_all or "gemini" in tools or "opencode" in tools or "pi" in tools
     # Claude's web-search server also needs a Responses-capable model.
     want_codex = (
         fetch_all or "codex" in tools or "claude" in tools or "copilot" in tools or "pi" in tools
     )
     # Codex smart routing can select OSS models such as GLM, so a Codex-only
     # configure must persist that discovered family too.
-    want_oss = fetch_all or "opencode" in tools or "codex" in tools
+    want_oss = fetch_all or "opencode" in tools or "kilo" in tools or "codex" in tools
 
     claude_reason: str | None = None
     gemini_reason: str | None = None
@@ -679,7 +686,8 @@ def configure_shared_state(
             state["codex_models"] = codex_models
         if want_oss:
             state["oss_models"] = oss_models
-        if fetch_all or "opencode" in tools:
+        if fetch_all or "opencode" in tools or "kilo" in tools:
+            # Kilo is an OpenCode fork and consumes the same discovered view.
             state["opencode_models"] = opencode_models
     save_state(state)
     # Scrub MCP entries that ucode wrote for the previous workspace so the new
@@ -3327,6 +3335,28 @@ def opencode_cmd(
     _launch_tool("opencode", ctx, model=model, skip_preflight=skip_preflight)
 
 
+@app.command(
+    "kilo",
+    context_settings={"allow_extra_args": True, "ignore_unknown_options": True},
+    rich_help_panel="Launch",
+)
+def kilo_cmd(
+    ctx: typer.Context,
+    model: Annotated[
+        str | None,
+        typer.Option(
+            "--model",
+            "-m",
+            help="Configured model ID or Kilo provider/model for this launch. "
+            "Pass before any `--` separator.",
+        ),
+    ] = None,
+    skip_preflight: SkipPreflightOption = False,
+) -> None:
+    """Launch Kilo via Databricks."""
+    _launch_tool("kilo", ctx, model=model, skip_preflight=skip_preflight)
+
+
 @app.command(
     "copilot",
     context_settings={"allow_extra_args": True, "ignore_unknown_options": True},
diff --git a/tests/test_agent_kilo.py b/tests/test_agent_kilo.py
new file mode 100644
index 000000000..30d5ac665
--- /dev/null
+++ b/tests/test_agent_kilo.py
@@ -0,0 +1,596 @@
+"""Tests for agents/kilo.py."""
+
+from __future__ import annotations
+
+import json
+from copy import deepcopy
+from unittest.mock import patch
+
+import pytest
+
+from ucode.agents import kilo
+from ucode.agents.args import LaunchOptions
+
+WS = "https://example.databricks.com"
+
+
+def _base_urls() -> dict[str, str]:
+    return {
+        "anthropic": f"{WS}/ai-gateway/anthropic/v1",
+        "gemini": f"{WS}/ai-gateway/gemini/v1beta",
+        "oss": f"{WS}/ai-gateway/mlflow/v1",
+    }
+
+
+class TestKiloSpec:
+    def test_binary(self):
+        assert kilo.SPEC["binary"] == "kilo"
+
+    def test_package(self):
+        assert kilo.SPEC["package"] == "@kilocode/cli@7"
+
+    def test_display(self):
+        assert kilo.SPEC["display"] == "Kilo"
+
+    def test_config_path(self):
+        assert kilo.SPEC["config_path"] == kilo.KILO_CONFIG_PATH
+
+    def test_requires_version_with_custom_provider_fetch(self, monkeypatch):
+        monkeypatch.setattr(kilo, "agent_version", lambda _binary: "7.3.0")
+
+        message = kilo.minimum_version_error()
+
+        assert message is not None
+        assert "requires Kilo 7.3.1 or newer" in message
+        assert "npm install -g @kilocode/cli@7" in message
+
+    def test_supported_version_needs_no_required_update(self, monkeypatch):
+        monkeypatch.setattr(kilo, "agent_version", lambda _binary: "7.3.1")
+
+        assert kilo.minimum_version_error() is None
+
+
+class TestAuthPlugin:
+    def test_calls_cross_platform_auth_token_helper_only_when_refreshing(self, monkeypatch):
+        monkeypatch.setattr("ucode.databricks.shutil.which", lambda command: f"/opt/{command}")
+
+        plugin = kilo.render_auth_plugin({"workspace": WS, "profile": "my profile"})
+
+        assert (
+            'const AUTH_COMMAND = ["/opt/ug", "auth-token", "--host", '
+            f'"{WS}", "--profile", "my profile", "--force-refresh"]'
+        ) in plugin
+        assert "run(AUTH_COMMAND[0], AUTH_COMMAND.slice(1)" in plugin
+        assert '"chat.headers"' not in plugin
+
+    def test_installs_cached_refreshing_fetch_on_databricks_providers(self):
+        plugin = kilo.render_auth_plugin({"workspace": WS})
+
+        assert "config: async (config)" in plugin
+        assert "options.fetch = databricksFetch" in plugin
+        assert "expiresAt <= Date.now() + REFRESH_SKEW_MS" in plugin
+        assert 'headers.set("Authorization", "Bearer " + token)' in plugin
+        assert "if (response.status !== 401) return response" in plugin
+        assert "return fetch(input, requestWithToken(input, init, accessToken))" in plugin
+
+    def test_refresh_is_single_flighted(self):
+        plugin = kilo.render_auth_plugin({"workspace": WS})
+
+        assert "if (!refreshPromise)" in plugin
+        assert "mintToken().finally(() => { refreshPromise = undefined })" in plugin
+
+
+class TestRenderOverlay:
+    def test_sets_model(self):
+        overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), {})
+        assert overlay["model"] == "claude-sonnet"
+
+    def test_anthropic_provider_added_when_models_present(self):
+        models = {"anthropic": ["claude-sonnet"], "gemini": []}
+        overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
+        assert "databricks-anthropic" in overlay["provider"]
+
+    def test_gemini_provider_added_when_models_present(self):
+        models = {"anthropic": [], "gemini": ["gemini-2"]}
+        overlay, _ = kilo.render_overlay("gemini-2", "tok", _base_urls(), models)
+        assert "databricks-google" in overlay["provider"]
+
+    def test_oss_provider_added_when_models_present(self):
+        models = {"oss": ["system.ai.kimi-k2-7-code"]}
+        overlay, _ = kilo.render_overlay("system.ai.kimi-k2-7-code", "tok", _base_urls(), models)
+        assert "databricks-oss" in overlay["provider"]
+
+    def test_oss_provider_uses_ai_sdk_openai_package(self):
+        models = {"oss": ["system.ai.kimi-k2-7-code"]}
+        overlay, _ = kilo.render_overlay("system.ai.kimi-k2-7-code", "tok", _base_urls(), models)
+        assert overlay["provider"]["databricks-oss"]["npm"] == "@ai-sdk/openai"
+
+    def test_deepseek_uses_oss_provider(self):
+        model = "system.ai.deepseek-v4-pro"
+
+        overlay, _ = kilo.render_overlay(model, "tok", _base_urls(), {"oss": [model]})
+
+        assert overlay["model"] == f"databricks-oss/{model}"
+        assert model in overlay["provider"]["databricks-oss"]["models"]
+
+    def test_both_providers_when_both_present(self):
+        models = {"anthropic": ["claude-sonnet"], "gemini": ["gemini-2"]}
+        overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
+        assert "databricks-anthropic" in overlay["provider"]
+        assert "databricks-google" in overlay["provider"]
+
+    def test_no_provider_key_when_no_models(self):
+        overlay, _ = kilo.render_overlay("model", "tok", _base_urls(), {})
+        assert "provider" not in overlay
+
+    def test_anthropic_base_url(self):
+        models = {"anthropic": ["claude-sonnet"]}
+        overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
+        options = overlay["provider"]["databricks-anthropic"]["options"]
+        assert options["baseURL"] == f"{WS}/ai-gateway/anthropic/v1"
+
+    def test_gemini_base_url(self):
+        models = {"gemini": ["gemini-2"]}
+        overlay, _ = kilo.render_overlay("gemini-2", "tok", _base_urls(), models)
+        options = overlay["provider"]["databricks-google"]["options"]
+        assert options["baseURL"] == f"{WS}/ai-gateway/gemini/v1beta"
+
+    def test_oss_base_url(self):
+        models = {"oss": ["system.ai.kimi-k2-7-code"]}
+        overlay, _ = kilo.render_overlay("system.ai.kimi-k2-7-code", "tok", _base_urls(), models)
+        options = overlay["provider"]["databricks-oss"]["options"]
+        assert options["baseURL"] == f"{WS}/ai-gateway/mlflow/v1"
+
+    def test_glm_gets_token_limits(self):
+        models = {"oss": ["system.ai.glm-5-2"]}
+        overlay, _ = kilo.render_overlay("system.ai.glm-5-2", "tok", _base_urls(), models)
+        glm = overlay["provider"]["databricks-oss"]["models"]["system.ai.glm-5-2"]
+        # Kilo's schema requires both context and output on `limit`.
+        assert glm["limit"] == {"context": 200000, "output": 25000}
+
+    def test_non_glm_oss_model_has_no_output_cap(self):
+        models = {"oss": ["system.ai.kimi-k2-7-code"]}
+        overlay, _ = kilo.render_overlay("system.ai.kimi-k2-7-code", "tok", _base_urls(), models)
+        kimi = overlay["provider"]["databricks-oss"]["models"]["system.ai.kimi-k2-7-code"]
+        assert "limit" not in kimi
+
+    def test_token_in_api_key(self):
+        models = {"anthropic": ["claude-sonnet"]}
+        overlay, _ = kilo.render_overlay("claude-sonnet", "mytoken", _base_urls(), models)
+        assert overlay["provider"]["databricks-anthropic"]["options"]["apiKey"] == "mytoken"
+
+    def test_authorization_header(self):
+        models = {"anthropic": ["claude-sonnet"]}
+        overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
+        headers = overlay["provider"]["databricks-anthropic"]["options"]["headers"]
+        assert headers["Authorization"] == "Bearer tok"
+
+    def test_anthropic_tool_streaming_disabled(self):
+        # @ai-sdk/anthropic injects `eager_input_streaming: true` on tool defs,
+        # which the Databricks gateway rejects. kilo's auto-disable skips
+        # Claude models, so we opt out per-model. The setting must live in
+        # `models.<m>.options` — per-call providerOptions — not provider options.
+        models = {"anthropic": ["claude-sonnet"]}
+        overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
+        model_entry = overlay["provider"]["databricks-anthropic"]["models"]["claude-sonnet"]
+        assert model_entry["options"]["toolStreaming"] is False
+
+    def test_user_agent_header_anthropic(self, monkeypatch):
+        # UA must live at the per-model level — Kilo clobbers
+        # provider-level `headers["User-Agent"]` in session/llm.ts.
+        monkeypatch.setattr(kilo, "ug_version", lambda: "0.1.0")
+        monkeypatch.setattr(kilo, "agent_version", lambda binary: "0.74.0")
+        models = {"anthropic": ["claude-sonnet"]}
+        overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
+        model_headers = overlay["provider"]["databricks-anthropic"]["models"]["claude-sonnet"][
+            "headers"
+        ]
+        assert model_headers["User-Agent"] == "ucode/0.1.0 kilo/0.74.0"
+
+    def test_user_agent_header_gemini(self, monkeypatch):
+        monkeypatch.setattr(kilo, "ug_version", lambda: "0.1.0")
+        monkeypatch.setattr(kilo, "agent_version", lambda binary: "0.74.0")
+        models = {"gemini": ["gemini-2"]}
+        overlay, _ = kilo.render_overlay("gemini-2", "tok", _base_urls(), models)
+        model_headers = overlay["provider"]["databricks-google"]["models"]["gemini-2"]["headers"]
+        assert model_headers["User-Agent"] == "ucode/0.1.0 kilo/0.74.0"
+
+    def test_provider_level_headers_only_authorization(self, monkeypatch):
+        # Sanity: provider-level headers should NOT include User-Agent (since
+        # it's clobbered there) — only Authorization.
+        models = {"anthropic": ["claude-sonnet"]}
+        overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
+        provider_headers = overlay["provider"]["databricks-anthropic"]["options"]["headers"]
+        assert "User-Agent" not in provider_headers
+        assert provider_headers["Authorization"] == "Bearer tok"
+
+    def test_managed_keys_include_model(self):
+        _, keys = kilo.render_overlay("model", "tok", _base_urls(), {})
+        assert ["model"] in keys
+
+    def test_managed_keys_include_anthropic_provider(self):
+        models = {"anthropic": ["claude-sonnet"]}
+        _, keys = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
+        assert ["provider", "databricks-anthropic"] in keys
+
+    def test_managed_keys_include_gemini_provider(self):
+        models = {"gemini": ["gemini-2"]}
+        _, keys = kilo.render_overlay("gemini-2", "tok", _base_urls(), models)
+        assert ["provider", "databricks-google"] in keys
+
+    def test_managed_keys_include_oss_provider(self):
+        models = {"oss": ["system.ai.kimi-k2-7-code"]}
+        _, keys = kilo.render_overlay("system.ai.kimi-k2-7-code", "tok", _base_urls(), models)
+        assert ["provider", "databricks-oss"] in keys
+
+    def test_anthropic_models_listed(self):
+        models = {"anthropic": ["claude-sonnet", "claude-haiku"]}
+        overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
+        provider_models = overlay["provider"]["databricks-anthropic"]["models"]
+        assert "claude-sonnet" in provider_models
+        assert "claude-haiku" in provider_models
+
+    def test_prefixes_anthropic_model_with_provider_id(self):
+        models = {"anthropic": ["claude-sonnet"], "gemini": []}
+        overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
+        assert overlay["model"] == "databricks-anthropic/claude-sonnet"
+
+    def test_prefixes_gemini_model_with_provider_id(self):
+        models = {"anthropic": [], "gemini": ["gemini-2"]}
+        overlay, _ = kilo.render_overlay("gemini-2", "tok", _base_urls(), models)
+        assert overlay["model"] == "databricks-google/gemini-2"
+
+    def test_prefixes_oss_model_with_provider_id(self):
+        models = {"oss": ["system.ai.kimi-k2-7-code"]}
+        overlay, _ = kilo.render_overlay("system.ai.kimi-k2-7-code", "tok", _base_urls(), models)
+        assert overlay["model"] == "databricks-oss/system.ai.kimi-k2-7-code"
+
+
+class TestMcpServerConfig:
+    # ucode registers the `ucode mcp-proxy ...` bridge as a `local` (stdio) MCP
+    # server; the proxy handles token refresh, so no URL/bearer header here.
+    PROXY_ARGV = ["ucode", "mcp-proxy", "--url", f"{WS}/api/2.0/mcp/functions/system/ai"]
+
+    def test_builds_local_server_entry_from_proxy_argv(self):
+        entry = kilo.build_mcp_server_entry(self.PROXY_ARGV)
+
+        assert entry == {
+            "type": "local",
+            "command": self.PROXY_ARGV,
+            "enabled": True,
+        }
+
+    def test_writes_mcp_server_without_clobbering_existing_config(self, tmp_path, monkeypatch):
+        import ucode.agents.kilo as oc_mod
+        import ucode.config_io as config_io_mod
+
+        monkeypatch.setattr(config_io_mod, "APP_DIR", tmp_path)
+        config_file = tmp_path / "kilo.json"
+        backup_file = tmp_path / "kilo-backup.json"
+        monkeypatch.setattr(oc_mod, "KILO_CONFIG_PATH", config_file)
+        monkeypatch.setattr(oc_mod, "KILO_BACKUP_PATH", backup_file)
+
+        config_file.write_text(
+            json.dumps(
+                {
+                    "model": "existing-model",
+                    "mcp": {"old-server": {"type": "local", "command": ["old"]}},
+                }
+            ),
+            encoding="utf-8",
+        )
+
+        removed = oc_mod.write_mcp_server_config("github", self.PROXY_ARGV)
+
+        written = json.loads(config_file.read_text())
+        assert removed is False
+        assert written["model"] == "existing-model"
+        assert written["mcp"]["old-server"] == {"type": "local", "command": ["old"]}
+        assert written["mcp"]["github"] == {
+            "type": "local",
+            "command": self.PROXY_ARGV,
+            "enabled": True,
+        }
+
+    def test_reports_replaced_mcp_server(self, tmp_path, monkeypatch):
+        import ucode.agents.kilo as oc_mod
+        import ucode.config_io as config_io_mod
+
+        monkeypatch.setattr(config_io_mod, "APP_DIR", tmp_path)
+        config_file = tmp_path / "kilo.json"
+        backup_file = tmp_path / "kilo-backup.json"
+        monkeypatch.setattr(oc_mod, "KILO_CONFIG_PATH", config_file)
+        monkeypatch.setattr(oc_mod, "KILO_BACKUP_PATH", backup_file)
+
+        config_file.write_text(json.dumps({"mcp": {"github": {"old": True}}}), encoding="utf-8")
+
+        removed = oc_mod.write_mcp_server_config("github", self.PROXY_ARGV)
+
+        assert removed is True
+        written = json.loads(config_file.read_text())
+        assert written["mcp"]["github"]["command"] == self.PROXY_ARGV
+
+    def test_removes_mcp_server_without_clobbering_others(self, tmp_path, monkeypatch):
+        import ucode.agents.kilo as oc_mod
+
+        config_file = tmp_path / "kilo.json"
+        monkeypatch.setattr(oc_mod, "KILO_CONFIG_PATH", config_file)
+        config_file.write_text(
+            json.dumps(
+                {
+                    "model": "existing-model",
+                    "mcp": {
+                        "github": {"url": "old"},
+                        "jira": {"url": "keep"},
+                    },
+                }
+            ),
+            encoding="utf-8",
+        )
+
+        removed = oc_mod.remove_mcp_server_config("github")
+
+        written = json.loads(config_file.read_text())
+        assert removed is True
+        assert "github" not in written["mcp"]
+        assert written["mcp"]["jira"] == {"url": "keep"}
+        assert written["model"] == "existing-model"
+
+
+class TestBuildRuntimeEnv:
+    def test_sets_oauth_token_for_mcp(self):
+        env = kilo.build_runtime_env("tok")
+
+        assert env["OAUTH_TOKEN"] == "tok"
+
+    def test_sets_kilo_config_isolation_vars(self):
+        # Kilo isolates via a single-file KILO_CONFIG (so a co-installed OpenCode
+        # config is never merged in) plus KILO_CONFIG_DIR for the plugin dir.
+        env = kilo.build_runtime_env("tok")
+
+        assert env["KILO_CONFIG"] == str(kilo.KILO_CONFIG_PATH)
+        assert env["KILO_CONFIG_DIR"] == str(kilo.KILO_CONFIG_INNER_DIR)
+
+
+class TestKiloDefaultModel:
+    def test_prefers_anthropic(self):
+        state = {"opencode_models": {"anthropic": ["claude-sonnet"], "gemini": ["gemini-2"]}}
+        assert kilo.default_model(state) == "claude-sonnet"
+
+    def test_falls_back_to_gemini(self):
+        state = {"opencode_models": {"anthropic": [], "gemini": ["gemini-2"]}}
+        assert kilo.default_model(state) == "gemini-2"
+
+    def test_falls_back_to_oss(self):
+        state = {
+            "opencode_models": {
+                "anthropic": [],
+                "gemini": [],
+                "oss": ["system.ai.kimi-k2-7-code"],
+            }
+        }
+        assert kilo.default_model(state) == "system.ai.kimi-k2-7-code"
+
+    def test_returns_none_when_empty(self):
+        assert kilo.default_model({}) is None
+        assert kilo.default_model({"opencode_models": {}}) is None
+
+    def test_kilo_default_model_wins_over_bucketed_models(self):
+        state = {
+            "opencode_default_model": "admin-chosen-default",
+            "opencode_models": {"anthropic": ["claude-sonnet"]},
+        }
+        assert kilo.default_model(state) == "admin-chosen-default"
+
+
+class TestKiloValidateCmd:
+    def test_starts_with_binary(self):
+        cmd = kilo.validate_cmd("kilo")
+        assert cmd[0] == "kilo"
+
+    def test_uses_run_subcommand(self):
+        cmd = kilo.validate_cmd("kilo")
+        assert "run" in cmd
+
+    def test_has_prompt(self):
+        cmd = kilo.validate_cmd("kilo")
+        assert len(cmd) > 2
+
+
+class TestKiloLaunchModel:
+    def test_resolves_configured_bare_and_native_selectors(self):
+        state = {
+            "opencode_models": {
+                "anthropic": ["claude-sonnet"],
+                "gemini": ["gemini-2"],
+                "oss": ["system.ai.kimi-k2-7-code"],
+            }
+        }
+
+        assert kilo.resolve_explicit_model("gemini-2", state) == "databricks-google/gemini-2"
+        assert (
+            kilo.resolve_explicit_model("openrouter/anthropic/claude-sonnet", state)
+            == "openrouter/anthropic/claude-sonnet"
+        )
+
+    def test_rejects_unknown_and_mismatched_managed_selectors(self):
+        state = {"opencode_models": {"anthropic": ["claude-sonnet"]}}
+
+        with pytest.raises(RuntimeError, match="not configured"):
+            kilo.resolve_explicit_model("missing-model", state)
+        with pytest.raises(RuntimeError, match="managed provider"):
+            kilo.resolve_explicit_model("databricks-google/claude-sonnet", state)
+
+    @pytest.mark.parametrize(
+        ("model", "tool_args", "selector", "expected_args"),
+        [
+            (
+                None,
+                ["run", "prompt"],
+                "databricks-anthropic/claude-sonnet",
+                ["run", "prompt"],
+            ),
+            (
+                "databricks-google/gemini-2",
+                ["run", "--", "--model", "literal"],
+                "databricks-google/gemini-2",
+                ["run", "--model", "databricks-google/gemini-2", "--", "--model", "literal"],
+            ),
+            (
+                "gemini-2",
+                ["run", "prompt"],
+                "databricks-google/gemini-2",
+                ["run", "prompt", "--model", "databricks-google/gemini-2"],
+            ),
+            (
+                "claude-sonnet",
+                ["run", "--model", "databricks-google/gemini-2"],
+                "databricks-google/gemini-2",
+                ["run", "--model", "databricks-google/gemini-2"],
+            ),
+            (
+                "openrouter/anthropic/claude-sonnet",
+                ["run", "--model", "openrouter/anthropic/claude-sonnet"],
+                "openrouter/anthropic/claude-sonnet",
+                ["run", "--model", "openrouter/anthropic/claude-sonnet"],
+            ),
+        ],
+    )
+    def test_launch_preserves_selection_and_saved_defaults(
+        self, tmp_path, monkeypatch, model, tool_args, selector, expected_args
+    ):
+        config_file = tmp_path / "kilo.json"
+        monkeypatch.setattr(kilo, "KILO_CONFIG_PATH", config_file)
+        monkeypatch.setattr(kilo, "KILO_BACKUP_PATH", tmp_path / "kilo-backup.json")
+        state = {
+            "workspace": WS,
+            "base_urls": {"kilo": _base_urls()},
+            "opencode_models": {"anthropic": ["claude-sonnet"], "gemini": ["gemini-2"]},
+            "opencode_default_model": "claude-sonnet",
+            "managed_configs": {},
+        }
+        original_state = deepcopy(state)
+        original_args = list(tool_args)
+        with (
+            patch("ucode.agents.kilo.get_databricks_token", return_value="tok"),
+            patch("ucode.agents.kilo.agent_version", return_value="1.0.220"),
+            patch("ucode.agents.kilo.save_state"),
+            patch("ucode.agents.kilo.subprocess_cross_os.popen") as popen,
+        ):
+            popen.return_value.wait.return_value = 7
+            with pytest.raises(SystemExit) as exc_info:
+                kilo.launch(state, tool_args, options=LaunchOptions(user_pinned_model=model))
+
+        assert exc_info.value.code == 7
+        assert json.loads(config_file.read_text())["model"] == selector
+        assert popen.call_args.args[0] == ["kilo", *expected_args]
+        assert tool_args == original_args
+        assert state["opencode_models"] == original_state["opencode_models"]
+        assert state["opencode_default_model"] == original_state["opencode_default_model"]
+
+    def test_rejects_invalid_model_before_configure_and_process(self):
+        state = {"opencode_models": {"anthropic": ["claude-sonnet"]}}
+        with (
+            patch("ucode.agents.kilo._configure_launch") as configure,
+            patch("ucode.agents.kilo.subprocess_cross_os.popen") as popen,
+            pytest.raises(RuntimeError, match="not configured"),
+        ):
+            kilo.launch(
+                state,
+                ["run", "--model", "missing-model"],
+                options=LaunchOptions(),
+            )
+
+        configure.assert_not_called()
+        popen.assert_not_called()
+
+
+class TestWriteToolConfigStaleProviderCleanup:
+    def test_stale_providers_removed_before_merge(self, tmp_path, monkeypatch):
+        import ucode.agents.kilo as oc_mod
+        import ucode.config_io as config_io_mod
+
+        monkeypatch.setattr(config_io_mod, "APP_DIR", tmp_path)
+        config_file = tmp_path / "kilo.json"
+        backup_file = tmp_path / "kilo-backup.json"
+        monkeypatch.setattr(oc_mod, "KILO_CONFIG_PATH", config_file)
+        monkeypatch.setattr(oc_mod, "KILO_BACKUP_PATH", backup_file)
+
+        stale = {
+            "provider": {
+                "databricks-anthropic": {"old": True},
+                "databricks-google": {"old": True},
+                "other-provider": {"keep": True},
+            }
+        }
+        config_file.write_text(json.dumps(stale), encoding="utf-8")
+
+        state = {
+            "workspace": WS,
+            "base_urls": {"kilo": _base_urls()},
+            "opencode_models": {"anthropic": ["claude-sonnet"]},
+            "managed_configs": {},
+        }
+
+        with (
+            patch("ucode.agents.kilo.get_databricks_token", return_value="tok"),
+            patch("ucode.agents.kilo.save_state"),
+        ):
+            oc_mod.write_tool_config(state, "claude-sonnet", token="tok")
+
+        written = json.loads(config_file.read_text())
+        providers = written.get("provider", {})
+        # stale entry is replaced with new data, not kept as-is
+        assert providers.get("databricks-anthropic") != {"old": True}
+        # unmanaged provider entry survives
+        assert providers.get("other-provider") == {"keep": True}
+        # Kilo 1.0.0 discovers `plugin/`; plural `plugins/` came later.
+        plugin = config_file.parent / "plugin" / kilo.KILO_AUTH_PLUGIN_PATH.name
+        assert plugin.exists()
+        assert "options.fetch = databricksFetch" in plugin.read_text()
+
+    def test_config_written_with_correct_model(self, tmp_path, monkeypatch):
+        import ucode.agents.kilo as oc_mod
+        import ucode.config_io as config_io_mod
+
+        monkeypatch.setattr(config_io_mod, "APP_DIR", tmp_path)
+        config_file = tmp_path / "kilo.json"
+        backup_file = tmp_path / "kilo-backup.json"
+        monkeypatch.setattr(oc_mod, "KILO_CONFIG_PATH", config_file)
+        monkeypatch.setattr(oc_mod, "KILO_BACKUP_PATH", backup_file)
+
+        state = {
+            "workspace": WS,
+            "base_urls": {"kilo": _base_urls()},
+            "opencode_models": {"anthropic": ["claude-sonnet"]},
+            "managed_configs": {},
+        }
+
+        with (
+            patch("ucode.agents.kilo.get_databricks_token", return_value="tok"),
+            patch("ucode.agents.kilo.save_state"),
+        ):
+            oc_mod.write_tool_config(state, "claude-sonnet", token="tok")
+
+        written = json.loads(config_file.read_text())
+        assert written["model"] == "databricks-anthropic/claude-sonnet"
+
+
+class TestWriteUserMcpServers:
+    def test_batched_add_remove_preserves_other_keys(self, tmp_path, monkeypatch):
+        path = tmp_path / "kilo.json"
+        path.write_text(
+            json.dumps({"provider": {"p": 1}, "mcp": {"mine": {"type": "local"}, "gone": {}}})
+        )
+        monkeypatch.setattr(kilo, "KILO_CONFIG_PATH", path)
+        monkeypatch.setattr(kilo, "KILO_BACKUP_PATH", tmp_path / "backup.json")
+
+        kilo.write_user_mcp_servers(
+            {"svc": kilo.build_mcp_server_entry(["ug", "mcp-proxy", "u"])}, {"gone"}
+        )
+
+        doc = json.loads(path.read_text())
+        assert doc["provider"] == {"p": 1}
+        assert "gone" not in doc["mcp"]
+        assert doc["mcp"]["mine"] == {"type": "local"}
+        assert doc["mcp"]["svc"]["command"] == ["ug", "mcp-proxy", "u"]
diff --git a/tests/test_agents_init.py b/tests/test_agents_init.py
index 2e1736c45..6cd92e871 100644
--- a/tests/test_agents_init.py
+++ b/tests/test_agents_init.py
@@ -54,7 +54,7 @@ def test_has_explicit_model_arg_stops_at_harness_separator(self):
 
 class TestToolSpecs:
     def test_all_tools_present(self):
-        assert set(TOOL_SPECS) == {"codex", "claude", "gemini", "opencode", "copilot", "pi"}
+        assert set(TOOL_SPECS) == {"codex", "claude", "gemini", "opencode", "kilo", "copilot", "pi"}
 
     def test_each_spec_has_required_keys(self):
         required = {"binary", "package", "display", "config_path", "backup_path"}
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 7e5124b8b..5575f1999 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -118,7 +118,7 @@ def test_help_groups_commands_by_workflow(self):
             "Manage": output[panels["Manage"] : panels["Usage"]],
             "Usage": output[panels["Usage"] : global_options],
         }
-        for command in ("claude", "codex", "copilot", "cursor", "gemini", "opencode", "pi"):
+        for command in ("claude", "codex", "copilot", "cursor", "gemini", "opencode", "kilo", "pi"):
             assert command in sections["Launch"]
         assert "configure" in sections["Setup"]
         for command in ("mcp", "skills"):
@@ -3384,7 +3384,9 @@ def test_agents_flag_rejects_unknown(self):
             result = runner.invoke(app, ["configure", "--agents", "claude,bogus"])
         assert result.exit_code != 0
         assert "Unsupported tool 'bogus'" in result.output
-        assert "codex, claude, gemini, opencode, copilot, pi" in " ".join(result.output.split())
+        assert "codex, claude, gemini, opencode, kilo, copilot, pi" in " ".join(
+            result.output.split()
+        )
         mock_cfg.assert_not_called()
 
     def test_agents_flag_rejects_empty_list(self):

From 45c82f91a5652de3db8e341114561bfd6ccaa047 Mon Sep 17 00:00:00 2001
From: Diego Jaramillo Celis <7494599+djcelis@users.noreply.github.com>
Date: Thu, 1 Oct 2026 15:05:55 -0400
Subject: [PATCH 2/2] Address review: isolate Kilo via XDG_CONFIG_HOME, wire +
 fix MCP, show models
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Isolation (core fix): switch Kilo config isolation from KILO_CONFIG to
XDG_CONFIG_HOME. Re-probing kilo 7.3.1 showed KILO_CONFIG is NOT exclusive —
it merely merges on top of the user's real ~/.config/kilo. XDG_CONFIG_HOME is
true isolation: Kilo reads only $XDG_CONFIG_HOME/kilo/, ignores the user's
real config, still discovers the plugin, and fires the config() hook (token
refresh works). Verified on 7.3.1/7.4.1.

MCP: wire Kilo into mcp.py (import, MCP_CLIENTS table, configure/remove
dispatch, _MCP_CLIENT_MODULES, and _managed_mcp_entry), mirroring OpenCode.
Live testing then surfaced two further gaps, now fixed:
  - `ug mcp list` ran `kilo mcp list` without the isolation env, so Kilo read
    the empty ~/.config/kilo and every server showed "missing". Added
    _kilo_cli_env() pinning XDG_CONFIG_HOME (mirrors _gemini_cli_env).
  - _parse_health_mcp_list did not understand Kilo's `●  ✓ name connected`
    tree output. Added a Kilo-scoped _parse_kilo_mcp_list. (The same gap
    affects OpenCode today; should be tracked as a separate follow-up.)
Verified end-to-end: `ug mcp add --agents kilo` registers servers in the
isolated config, `ug mcp list --agents kilo` reports connected/failed
status, and a `ug kilo run` invoked system.ai.dbsql.execute_sql successfully.

Status: _status_models/_status_default_model surface Kilo's models,
which it shares with OpenCode under opencode_models.

Cleanups:  collapse the duplicate KILO_CONFIG_INNER_DIR constant; remove the
dead opencode_default_model branch in default_model(); add an actionable hint
when no models are discovered; add Kilo to the --agent help text, help
command order, and token-refresh launch note. Remove inaccurate comments;
correct the version-floor comment.

Tests: assert XDG isolation (XDG_CONFIG_HOME set; KILO_CONFIG not set), add
Kilo MCP dispatch + status-env + tree-parse coverage, fix the shared
base_urls key to "opencode", use 7.3.1 version, drop the oc_mod
alias, and correct the default-model test to the bucketed-derivation behavior.
---
 src/ucode/agents/kilo.py | 65 ++++++++++------------------
 src/ucode/cli.py         | 10 +++--
 src/ucode/mcp.py         | 64 ++++++++++++++++++++++++++--
 tests/test_agent_kilo.py | 86 +++++++++++++++++--------------------
 tests/test_mcp.py        | 92 ++++++++++++++++++++++++++++++++++++++++
 5 files changed, 222 insertions(+), 95 deletions(-)

diff --git a/src/ucode/agents/kilo.py b/src/ucode/agents/kilo.py
index 4dca77d1f..053554ac5 100644
--- a/src/ucode/agents/kilo.py
+++ b/src/ucode/agents/kilo.py
@@ -3,8 +3,8 @@
 Kilo CLI (@kilocode/cli) is a fork of OpenCode; it shares OpenCode's config
 schema, provider model, and plugin system. This module mirrors opencode.py,
 reading the same discovered `opencode_models` view from state, and isolating
-Kilo via a single-file KILO_CONFIG (proven exclusive on kilo 7.2.x, whereas
-KILO_CONFIG_DIR merges the user's real config dir underneath)."""
+Kilo via XDG_CONFIG_HOME (verified on kilo 7.3.1/7.4.1: Kilo reads only
+$XDG_CONFIG_HOME/kilo/ and ignores the user's real ~/.config/kilo)."""
 
 from __future__ import annotations
 
@@ -35,19 +35,13 @@
 
 from .args import LaunchOptions, explicit_model_arg_value, has_explicit_model_arg
 
-KILO_CONFIG_DIR = APP_DIR / "kilo-config"
-KILO_CONFIG_INNER_DIR = KILO_CONFIG_DIR
-KILO_CONFIG_PATH = KILO_CONFIG_INNER_DIR / "kilo.json"
+KILO_XDG_CONFIG_HOME = APP_DIR / "kilo-xdg"
+KILO_CONFIG_DIR = KILO_XDG_CONFIG_HOME / "kilo"
+KILO_CONFIG_PATH = KILO_CONFIG_DIR / "kilo.json"
 KILO_BACKUP_PATH = APP_DIR / "kilo-config.backup.json"
-KILO_AUTH_PLUGIN_PATH = KILO_CONFIG_INNER_DIR / "plugin" / "ucode-auth.js"
+KILO_AUTH_PLUGIN_PATH = KILO_CONFIG_DIR / "plugin" / "ucode-auth.js"
 KILO_NPM_PACKAGE = "@kilocode/cli"
-# Minimum Kilo version whose plugin runtime invokes the `config()` hook that our
-# auth plugin (ucode-auth.js) relies on to refresh Databricks tokens.
-# Verified empirically (plugin `config()` probe):
-#   7.2.1  -> hook does NOT fire (unsupported; 7.2.x is a noisy patch series)
-#   7.2.25 -> hook fires
-#   7.3.1  -> hook fires  <-- chosen floor: earliest clean minor boundary that works
-#   7.4.1  -> hook fires
+# Minimum Kilo version based on tests is 7.3.1
 MINIMUM_KILO_VERSION = (7, 3, 1)
 MINIMUM_KILO_VERSION_TEXT = "7.3.1"
 
@@ -220,10 +214,13 @@ def resolve_explicit_model(model: str, state: dict) -> str:
     if "/" not in selector:
         if resolved != selector:
             return resolved
-        raise RuntimeError(
-            f"Kilo model '{selector}' is not configured. "
-            "Choose a discovered model id or pass a provider/model selector."
+        hint = (
+            "Run `ug configure --agent kilo` to discover models, then choose a "
+            "discovered model id or pass a provider/model selector."
+            if not opencode_models
+            else "Choose a discovered model id or pass a provider/model selector."
         )
+        raise RuntimeError(f"Kilo model '{selector}' is not configured. {hint}")
 
     provider, _, model_id = selector.partition("/")
     if not provider or not model_id:
@@ -240,12 +237,7 @@ def resolve_explicit_model(model: str, state: dict) -> str:
 
 
 def _oss_model_overlay(model: str, ua_header: dict[str, str]) -> dict:
-    """Per-model overlay for an OSS model entry.
-
-    All OSS models carry the User-Agent header; models with known token limits
-    also pin `limit` (context + output) so Kilo clamps `max_tokens` to a
-    value the gateway accepts. Kilo's schema requires both fields together,
-    so the limits table always supplies both."""
+    """Per-model overlay for an OSS model entry."""
     overlay: dict = {"headers": ua_header}
     limits = model_token_limits(model)
     if limits is not None:
@@ -261,10 +253,8 @@ def render_overlay(
 ) -> tuple[dict, list[list[str]]]:
     """Return (overlay, managed_key_paths) for kilo.json."""
     auth_headers = {"Authorization": f"Bearer {token}"}
-    # Kilo hardcodes `User-Agent: opencode/<ver>` in session/llm.ts for
-    # every provider, after the AI SDK's combineHeaders. The provider-level
-    # `headers` are clobbered by that injection, but per-model `headers` are
-    # merged AFTER and win — so the UA must live on each model entry.
+    # The User-Agent must be set on each per-model `headers` entry; a
+    # provider-level `headers["User-Agent"]` does not take effect.
     ua_header = {
         "User-Agent": f"ucode/{ug_version()} kilo/{agent_version('kilo')}",
     }
@@ -276,11 +266,6 @@ def render_overlay(
     providers: dict = {}
     keys: list[list[str]] = [["model"]]
     if anthropic_models:
-        # @ai-sdk/anthropic injects `eager_input_streaming: true` on tool defs;
-        # the Databricks gateway's strict validator rejects it. opencode's
-        # auto-disable in transform.ts skips models whose id contains "claude",
-        # so we opt out per-model. The setting lives in per-call providerOptions,
-        # which opencode reads from `models.<m>.options`, not provider `options`.
         anthropic_model_overlay = {
             "headers": ua_header,
             "options": {"toolStreaming": False},
@@ -348,7 +333,6 @@ def write_tool_config(
         for stale in (
             "databricks-anthropic",
             "databricks-google",
-            "databricks-openai",
             "databricks-oss",
         ):
             providers.pop(stale, None)
@@ -400,8 +384,6 @@ def write_user_mcp_servers(add: dict[str, dict], remove: set[str]) -> set[str]:
 
 
 def default_model(state: dict) -> str | None:
-    if isinstance(state.get("opencode_default_model"), str):
-        return state.get("opencode_default_model")
     opencode_models = state.get("opencode_models") or {}
     anthropic = opencode_models.get("anthropic") or []
     if anthropic:
@@ -424,14 +406,13 @@ def _configure_launch(state: dict, model: str | None = None) -> str:
 def build_runtime_env(token: str, state: dict | None = None) -> dict[str, str]:
     env = os.environ.copy()
     env["OAUTH_TOKEN"] = token
-    # Isolation: unlike OpenCode (which neutralizes XDG_CONFIG_HOME), Kilo pins a
-    # single config file via KILO_CONFIG. Verified empirically against kilo 7.2.25:
-    # with KILO_CONFIG=<file> set, Kilo loads ONLY that file and does NOT merge the
-    # user's real ~/.config/kilo. (KILO_CONFIG_DIR alone is additive/merges the real
-    # dir, so it is used only to point Kilo at our plugin directory, not for config
-    # isolation.) This keeps a co-installed OpenCode's config out of Kilo's merge chain.
-    env["KILO_CONFIG"] = str(KILO_CONFIG_PATH)
-    env["KILO_CONFIG_DIR"] = str(KILO_CONFIG_INNER_DIR)
+    # Isolation: like OpenCode, Kilo honors XDG_CONFIG_HOME. Pointing it at our
+    # private dir makes Kilo read ONLY $XDG_CONFIG_HOME/kilo/ and ignore the user's
+    # real ~/.config/kilo. Verified on kilo 7.3.1/7.4.1: config() plugin hook fires
+    # (token refresh works) and `kilo mcp list` sees only our managed servers.
+    # Note: KILO_CONFIG merely MERGES on top of the real config (not exclusive), so
+    # it is not used for isolation.
+    env["XDG_CONFIG_HOME"] = str(KILO_XDG_CONFIG_HOME)
     return env
 
 
diff --git a/src/ucode/cli.py b/src/ucode/cli.py
index 3aa3baaea..c4797d4f7 100644
--- a/src/ucode/cli.py
+++ b/src/ucode/cli.py
@@ -1070,6 +1070,9 @@ def _status_models(tool: str, state: dict) -> list[str]:
         models = static_models
     elif tool in ("claude", "codex", "gemini", "opencode"):
         models = _model_values(state.get(f"{tool}_models"))
+    elif tool == "kilo":
+        # Kilo shares OpenCode's discovery; its models live under opencode_models.
+        models = _model_values(state.get("opencode_models"))
     elif tool == "copilot":
         models = _model_values(state.get("copilot_models")) or (
             _model_values(state.get("claude_models")) + _model_values(state.get("codex_models"))
@@ -1091,7 +1094,7 @@ def _status_default_model(tool: str, state: dict, models: list[str]) -> str | No
         return explicit
     # Claude and Codex deliberately leave the starting model to the agent unless a managed
     # config pins one. The other clients write the first resolved model into their ug config.
-    return models[0] if models and tool in ("gemini", "opencode", "copilot", "pi") else None
+    return models[0] if models and tool in ("gemini", "opencode", "kilo", "copilot", "pi") else None
 
 
 def _live_status_model_state(state: dict, tools: set[str]) -> tuple[dict, str]:
@@ -1318,6 +1321,7 @@ def revert() -> int:
     "cursor",
     "gemini",
     "opencode",
+    "kilo",
     "pi",
     "configure",
     "mcp",
@@ -2914,7 +2918,7 @@ def _launch_tool(
                 f"{TOOL_SPECS[tool]['display']} may require one-time hook review. Open "
                 "`/hooks` and trust the ug routing hooks if prompted."
             )
-        if tool in ("gemini", "opencode", "copilot", "pi"):
+        if tool in ("gemini", "opencode", "kilo", "copilot", "pi"):
             print_note(
                 f"{TOOL_SPECS[tool]['display']} token refresh is managed automatically "
                 f"every 30 minutes while the session is running."
@@ -3430,7 +3434,7 @@ def configure(
         str | None,
         typer.Option(
             "--agent",
-            help="Configure only the named agent (e.g. claude, codex, gemini, opencode, copilot, pi).",
+            help="Configure only the named agent (e.g. claude, codex, gemini, opencode, kilo, copilot, pi).",
         ),
     ] = None,
     agents: Annotated[
diff --git a/src/ucode/mcp.py b/src/ucode/mcp.py
index fd4eaab09..b447add1c 100644
--- a/src/ucode/mcp.py
+++ b/src/ucode/mcp.py
@@ -15,7 +15,7 @@
 import questionary
 from rich.table import Table
 
-from ucode.agents import claude, codex, copilot, cursor, gemini, opencode
+from ucode.agents import claude, codex, copilot, cursor, gemini, kilo, opencode
 from ucode.config_io import restore_file
 from ucode.constants import MCP_CLEANUP_SCOPES, MCP_USER_SCOPE
 from ucode.databricks import (
@@ -116,6 +116,11 @@
         "display": "OpenCode",
         "list_command": "opencode mcp list",
     },
+    "kilo": {
+        "binary": "kilo",
+        "display": "Kilo",
+        "list_command": "kilo mcp list",
+    },
     "copilot": {
         "binary": "copilot",
         "display": "GitHub Copilot CLI",
@@ -226,6 +231,14 @@ def _gemini_cli_env() -> dict[str, str]:
     return env
 
 
+def _kilo_cli_env() -> dict[str, str]:
+    # Pin XDG_CONFIG_HOME to the isolated dir ug writes Kilo's config into, so
+    # `kilo mcp list` reads the same config (matches kilo.build_runtime_env).
+    env = os.environ.copy()
+    env["XDG_CONFIG_HOME"] = str(kilo.KILO_XDG_CONFIG_HOME)
+    return env
+
+
 def add_gemini_mcp_server(name: str, argv: list[str]) -> None:
     # Register the proxy as a stdio server: `gemini mcp add <name> <cmd> <args…>
     # --type stdio`. The scope/type flags trail the captured command + args.
@@ -389,6 +402,9 @@ def configure_client_mcp_server(
     if client == "opencode":
         removed = opencode.write_mcp_server_config(name, argv)
         return [MCP_USER_SCOPE] if removed else []
+    if client == "kilo":
+        removed = kilo.write_mcp_server_config(name, argv)
+        return [MCP_USER_SCOPE] if removed else []
     if client == "copilot":
         removed = copilot.write_mcp_server_config(name, argv)
         return [MCP_USER_SCOPE] if removed else []
@@ -409,6 +425,8 @@ def remove_client_mcp_server(client: str, name: str) -> list[str]:
         return [MCP_USER_SCOPE] if remove_gemini_mcp_server(name) else []
     if client == "opencode":
         return [MCP_USER_SCOPE] if opencode.remove_mcp_server_config(name) else []
+    if client == "kilo":
+        return [MCP_USER_SCOPE] if kilo.remove_mcp_server_config(name) else []
     if client == "copilot":
         return [MCP_USER_SCOPE] if copilot.remove_mcp_server_config(name) else []
     if client == "cursor":
@@ -1348,6 +1366,7 @@ def _discover_selected_mcp_sources(
     "copilot": copilot,
     "cursor": cursor,
     "opencode": opencode,
+    "kilo": kilo,
 }
 
 
@@ -1391,6 +1410,8 @@ def _managed_mcp_entry(
         return copilot.build_mcp_server_entry(argv)
     if client == "opencode":
         return opencode.build_mcp_server_entry(argv)
+    if client == "kilo":
+        return kilo.build_mcp_server_entry(argv)
     raise RuntimeError(f"Unsupported MCP client '{client}'.")
 
 
@@ -2333,6 +2354,31 @@ def _parse_codex_mcp_list(output: str) -> dict[str, str]:
     return statuses
 
 
+def _parse_kilo_mcp_list(output: str) -> dict[str, str]:
+    """Parse `kilo mcp list`'s tree output into ``{server_name: state}``.
+
+    Kilo prints one server per line as ``●  <glyph> <name> <status>`` (glyph ✓/✗), with indented
+    ``│``-prefixed detail lines (the proxy command, error text) that must be skipped so the command
+    URL isn't misread as a server. (This is the OpenCode-family shape; parsing it generically for
+    every client is tracked separately — here it is scoped to Kilo.)
+    """
+    statuses: dict[str, str] = {}
+    for raw in output.splitlines():
+        line = raw.strip()
+        # Server rows start with the ● bullet; ┌/│/└ tree lines are headers/details.
+        if not line.startswith("●"):
+            continue
+        tokens = line.lstrip("● ").split()
+        # Expect: <glyph> <name> <status…>; require the health glyph so headers don't slip through.
+        if len(tokens) < 2 or tokens[0] not in (_HEALTH_OK_MARKERS + _HEALTH_FAIL_MARKERS):
+            continue
+        name = tokens[1]
+        if " " in name:
+            continue
+        statuses[name] = _classify_health_line(line)
+    return statuses
+
+
 def parse_mcp_list_output(client: str, output: str) -> dict[str, str]:
     """Parse an agent's `mcp list` output into ``{server_name: live-state}`` (best-effort).
 
@@ -2341,7 +2387,12 @@ def parse_mcp_list_output(client: str, output: str) -> dict[str, str]:
     failure detail (``Failed to connect - HTTP 404 Not Found``), so checking it up front would
     let one broken server empty the whole listing.
     """
-    parsed = _parse_codex_mcp_list(output) if client == "codex" else _parse_health_mcp_list(output)
+    if client == "codex":
+        parsed = _parse_codex_mcp_list(output)
+    elif client == "kilo":
+        parsed = _parse_kilo_mcp_list(output)
+    else:
+        parsed = _parse_health_mcp_list(output)
     if not parsed and _is_missing_mcp_server_output(output):
         return {}
     return parsed
@@ -2357,8 +2408,13 @@ def _run_mcp_list(client: str) -> str | None:
     if not spec:
         return None
     argv = str(spec["list_command"]).split()
-    # Gemini reads its config from a pinned home dir, matching how ucode registers servers there.
-    env = _gemini_cli_env() if client == "gemini" else None
+    # Gemini and Kilo read their config from a pinned dir, matching how ucode registers servers there.
+    if client == "gemini":
+        env = _gemini_cli_env()
+    elif client == "kilo":
+        env = _kilo_cli_env()
+    else:
+        env = None
     try:
         result = subprocess_cross_os.run(
             argv,
diff --git a/tests/test_agent_kilo.py b/tests/test_agent_kilo.py
index 30d5ac665..8b349ac7a 100644
--- a/tests/test_agent_kilo.py
+++ b/tests/test_agent_kilo.py
@@ -145,7 +145,6 @@ def test_glm_gets_token_limits(self):
         models = {"oss": ["system.ai.glm-5-2"]}
         overlay, _ = kilo.render_overlay("system.ai.glm-5-2", "tok", _base_urls(), models)
         glm = overlay["provider"]["databricks-oss"]["models"]["system.ai.glm-5-2"]
-        # Kilo's schema requires both context and output on `limit`.
         assert glm["limit"] == {"context": 200000, "output": 25000}
 
     def test_non_glm_oss_model_has_no_output_cap(self):
@@ -166,38 +165,30 @@ def test_authorization_header(self):
         assert headers["Authorization"] == "Bearer tok"
 
     def test_anthropic_tool_streaming_disabled(self):
-        # @ai-sdk/anthropic injects `eager_input_streaming: true` on tool defs,
-        # which the Databricks gateway rejects. kilo's auto-disable skips
-        # Claude models, so we opt out per-model. The setting must live in
-        # `models.<m>.options` — per-call providerOptions — not provider options.
         models = {"anthropic": ["claude-sonnet"]}
         overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
         model_entry = overlay["provider"]["databricks-anthropic"]["models"]["claude-sonnet"]
         assert model_entry["options"]["toolStreaming"] is False
 
     def test_user_agent_header_anthropic(self, monkeypatch):
-        # UA must live at the per-model level — Kilo clobbers
-        # provider-level `headers["User-Agent"]` in session/llm.ts.
         monkeypatch.setattr(kilo, "ug_version", lambda: "0.1.0")
-        monkeypatch.setattr(kilo, "agent_version", lambda binary: "0.74.0")
+        monkeypatch.setattr(kilo, "agent_version", lambda binary: "7.3.1")
         models = {"anthropic": ["claude-sonnet"]}
         overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
         model_headers = overlay["provider"]["databricks-anthropic"]["models"]["claude-sonnet"][
             "headers"
         ]
-        assert model_headers["User-Agent"] == "ucode/0.1.0 kilo/0.74.0"
+        assert model_headers["User-Agent"] == "ucode/0.1.0 kilo/7.3.1"
 
     def test_user_agent_header_gemini(self, monkeypatch):
         monkeypatch.setattr(kilo, "ug_version", lambda: "0.1.0")
-        monkeypatch.setattr(kilo, "agent_version", lambda binary: "0.74.0")
+        monkeypatch.setattr(kilo, "agent_version", lambda binary: "7.3.1")
         models = {"gemini": ["gemini-2"]}
         overlay, _ = kilo.render_overlay("gemini-2", "tok", _base_urls(), models)
         model_headers = overlay["provider"]["databricks-google"]["models"]["gemini-2"]["headers"]
-        assert model_headers["User-Agent"] == "ucode/0.1.0 kilo/0.74.0"
+        assert model_headers["User-Agent"] == "ucode/0.1.0 kilo/7.3.1"
 
     def test_provider_level_headers_only_authorization(self, monkeypatch):
-        # Sanity: provider-level headers should NOT include User-Agent (since
-        # it's clobbered there) — only Authorization.
         models = {"anthropic": ["claude-sonnet"]}
         overlay, _ = kilo.render_overlay("claude-sonnet", "tok", _base_urls(), models)
         provider_headers = overlay["provider"]["databricks-anthropic"]["options"]["headers"]
@@ -261,14 +252,13 @@ def test_builds_local_server_entry_from_proxy_argv(self):
         }
 
     def test_writes_mcp_server_without_clobbering_existing_config(self, tmp_path, monkeypatch):
-        import ucode.agents.kilo as oc_mod
         import ucode.config_io as config_io_mod
 
         monkeypatch.setattr(config_io_mod, "APP_DIR", tmp_path)
         config_file = tmp_path / "kilo.json"
         backup_file = tmp_path / "kilo-backup.json"
-        monkeypatch.setattr(oc_mod, "KILO_CONFIG_PATH", config_file)
-        monkeypatch.setattr(oc_mod, "KILO_BACKUP_PATH", backup_file)
+        monkeypatch.setattr(kilo, "KILO_CONFIG_PATH", config_file)
+        monkeypatch.setattr(kilo, "KILO_BACKUP_PATH", backup_file)
 
         config_file.write_text(
             json.dumps(
@@ -280,7 +270,7 @@ def test_writes_mcp_server_without_clobbering_existing_config(self, tmp_path, mo
             encoding="utf-8",
         )
 
-        removed = oc_mod.write_mcp_server_config("github", self.PROXY_ARGV)
+        removed = kilo.write_mcp_server_config("github", self.PROXY_ARGV)
 
         written = json.loads(config_file.read_text())
         assert removed is False
@@ -293,28 +283,26 @@ def test_writes_mcp_server_without_clobbering_existing_config(self, tmp_path, mo
         }
 
     def test_reports_replaced_mcp_server(self, tmp_path, monkeypatch):
-        import ucode.agents.kilo as oc_mod
         import ucode.config_io as config_io_mod
 
         monkeypatch.setattr(config_io_mod, "APP_DIR", tmp_path)
         config_file = tmp_path / "kilo.json"
         backup_file = tmp_path / "kilo-backup.json"
-        monkeypatch.setattr(oc_mod, "KILO_CONFIG_PATH", config_file)
-        monkeypatch.setattr(oc_mod, "KILO_BACKUP_PATH", backup_file)
+        monkeypatch.setattr(kilo, "KILO_CONFIG_PATH", config_file)
+        monkeypatch.setattr(kilo, "KILO_BACKUP_PATH", backup_file)
 
         config_file.write_text(json.dumps({"mcp": {"github": {"old": True}}}), encoding="utf-8")
 
-        removed = oc_mod.write_mcp_server_config("github", self.PROXY_ARGV)
+        removed = kilo.write_mcp_server_config("github", self.PROXY_ARGV)
 
         assert removed is True
         written = json.loads(config_file.read_text())
         assert written["mcp"]["github"]["command"] == self.PROXY_ARGV
 
     def test_removes_mcp_server_without_clobbering_others(self, tmp_path, monkeypatch):
-        import ucode.agents.kilo as oc_mod
 
         config_file = tmp_path / "kilo.json"
-        monkeypatch.setattr(oc_mod, "KILO_CONFIG_PATH", config_file)
+        monkeypatch.setattr(kilo, "KILO_CONFIG_PATH", config_file)
         config_file.write_text(
             json.dumps(
                 {
@@ -328,7 +316,7 @@ def test_removes_mcp_server_without_clobbering_others(self, tmp_path, monkeypatc
             encoding="utf-8",
         )
 
-        removed = oc_mod.remove_mcp_server_config("github")
+        removed = kilo.remove_mcp_server_config("github")
 
         written = json.loads(config_file.read_text())
         assert removed is True
@@ -343,13 +331,20 @@ def test_sets_oauth_token_for_mcp(self):
 
         assert env["OAUTH_TOKEN"] == "tok"
 
-    def test_sets_kilo_config_isolation_vars(self):
-        # Kilo isolates via a single-file KILO_CONFIG (so a co-installed OpenCode
-        # config is never merged in) plus KILO_CONFIG_DIR for the plugin dir.
+    def test_isolates_via_xdg_config_home(self):
+        # Kilo isolates via XDG_CONFIG_HOME: it reads ONLY $XDG_CONFIG_HOME/kilo/
+        # and ignores the user's real ~/.config/kilo. KILO_CONFIG merely merges on
+        # top of the real config (not exclusive on 7.3.1), so it must not be set.
         env = kilo.build_runtime_env("tok")
 
-        assert env["KILO_CONFIG"] == str(kilo.KILO_CONFIG_PATH)
-        assert env["KILO_CONFIG_DIR"] == str(kilo.KILO_CONFIG_INNER_DIR)
+        assert env["XDG_CONFIG_HOME"] == str(kilo.KILO_XDG_CONFIG_HOME)
+        assert "KILO_CONFIG" not in env
+        assert "KILO_CONFIG_DIR" not in env
+
+    def test_config_path_lives_under_xdg_home(self):
+        # The config dir Kilo reads is $XDG_CONFIG_HOME/kilo, and kilo.json sits in it.
+        assert kilo.KILO_CONFIG_DIR == kilo.KILO_XDG_CONFIG_HOME / "kilo"
+        assert kilo.KILO_CONFIG_PATH == kilo.KILO_CONFIG_DIR / "kilo.json"
 
 
 class TestKiloDefaultModel:
@@ -375,12 +370,14 @@ def test_returns_none_when_empty(self):
         assert kilo.default_model({}) is None
         assert kilo.default_model({"opencode_models": {}}) is None
 
-    def test_kilo_default_model_wins_over_bucketed_models(self):
+    def test_default_derives_from_bucketed_models(self):
+        # Kilo has no `kilo_default_model` key and ignores `opencode_default_model`;
+        # the default is the first bucketed model (anthropic > gemini > oss).
         state = {
-            "opencode_default_model": "admin-chosen-default",
-            "opencode_models": {"anthropic": ["claude-sonnet"]},
+            "opencode_default_model": "ignored-key",
+            "opencode_models": {"anthropic": ["claude-sonnet"], "gemini": ["gemini-2"]},
         }
-        assert kilo.default_model(state) == "admin-chosen-default"
+        assert kilo.default_model(state) == "claude-sonnet"
 
 
 class TestKiloValidateCmd:
@@ -464,7 +461,7 @@ def test_launch_preserves_selection_and_saved_defaults(
         monkeypatch.setattr(kilo, "KILO_BACKUP_PATH", tmp_path / "kilo-backup.json")
         state = {
             "workspace": WS,
-            "base_urls": {"kilo": _base_urls()},
+            "base_urls": {"opencode": _base_urls()},
             "opencode_models": {"anthropic": ["claude-sonnet"], "gemini": ["gemini-2"]},
             "opencode_default_model": "claude-sonnet",
             "managed_configs": {},
@@ -473,7 +470,7 @@ def test_launch_preserves_selection_and_saved_defaults(
         original_args = list(tool_args)
         with (
             patch("ucode.agents.kilo.get_databricks_token", return_value="tok"),
-            patch("ucode.agents.kilo.agent_version", return_value="1.0.220"),
+            patch("ucode.agents.kilo.agent_version", return_value="7.3.1"),
             patch("ucode.agents.kilo.save_state"),
             patch("ucode.agents.kilo.subprocess_cross_os.popen") as popen,
         ):
@@ -507,14 +504,13 @@ def test_rejects_invalid_model_before_configure_and_process(self):
 
 class TestWriteToolConfigStaleProviderCleanup:
     def test_stale_providers_removed_before_merge(self, tmp_path, monkeypatch):
-        import ucode.agents.kilo as oc_mod
         import ucode.config_io as config_io_mod
 
         monkeypatch.setattr(config_io_mod, "APP_DIR", tmp_path)
         config_file = tmp_path / "kilo.json"
         backup_file = tmp_path / "kilo-backup.json"
-        monkeypatch.setattr(oc_mod, "KILO_CONFIG_PATH", config_file)
-        monkeypatch.setattr(oc_mod, "KILO_BACKUP_PATH", backup_file)
+        monkeypatch.setattr(kilo, "KILO_CONFIG_PATH", config_file)
+        monkeypatch.setattr(kilo, "KILO_BACKUP_PATH", backup_file)
 
         stale = {
             "provider": {
@@ -527,7 +523,7 @@ def test_stale_providers_removed_before_merge(self, tmp_path, monkeypatch):
 
         state = {
             "workspace": WS,
-            "base_urls": {"kilo": _base_urls()},
+            "base_urls": {"opencode": _base_urls()},
             "opencode_models": {"anthropic": ["claude-sonnet"]},
             "managed_configs": {},
         }
@@ -536,7 +532,7 @@ def test_stale_providers_removed_before_merge(self, tmp_path, monkeypatch):
             patch("ucode.agents.kilo.get_databricks_token", return_value="tok"),
             patch("ucode.agents.kilo.save_state"),
         ):
-            oc_mod.write_tool_config(state, "claude-sonnet", token="tok")
+            kilo.write_tool_config(state, "claude-sonnet", token="tok")
 
         written = json.loads(config_file.read_text())
         providers = written.get("provider", {})
@@ -544,24 +540,22 @@ def test_stale_providers_removed_before_merge(self, tmp_path, monkeypatch):
         assert providers.get("databricks-anthropic") != {"old": True}
         # unmanaged provider entry survives
         assert providers.get("other-provider") == {"keep": True}
-        # Kilo 1.0.0 discovers `plugin/`; plural `plugins/` came later.
         plugin = config_file.parent / "plugin" / kilo.KILO_AUTH_PLUGIN_PATH.name
         assert plugin.exists()
         assert "options.fetch = databricksFetch" in plugin.read_text()
 
     def test_config_written_with_correct_model(self, tmp_path, monkeypatch):
-        import ucode.agents.kilo as oc_mod
         import ucode.config_io as config_io_mod
 
         monkeypatch.setattr(config_io_mod, "APP_DIR", tmp_path)
         config_file = tmp_path / "kilo.json"
         backup_file = tmp_path / "kilo-backup.json"
-        monkeypatch.setattr(oc_mod, "KILO_CONFIG_PATH", config_file)
-        monkeypatch.setattr(oc_mod, "KILO_BACKUP_PATH", backup_file)
+        monkeypatch.setattr(kilo, "KILO_CONFIG_PATH", config_file)
+        monkeypatch.setattr(kilo, "KILO_BACKUP_PATH", backup_file)
 
         state = {
             "workspace": WS,
-            "base_urls": {"kilo": _base_urls()},
+            "base_urls": {"opencode": _base_urls()},
             "opencode_models": {"anthropic": ["claude-sonnet"]},
             "managed_configs": {},
         }
@@ -570,7 +564,7 @@ def test_config_written_with_correct_model(self, tmp_path, monkeypatch):
             patch("ucode.agents.kilo.get_databricks_token", return_value="tok"),
             patch("ucode.agents.kilo.save_state"),
         ):
-            oc_mod.write_tool_config(state, "claude-sonnet", token="tok")
+            kilo.write_tool_config(state, "claude-sonnet", token="tok")
 
         written = json.loads(config_file.read_text())
         assert written["model"] == "databricks-anthropic/claude-sonnet"
diff --git a/tests/test_mcp.py b/tests/test_mcp.py
index 85b5ddcb8..4d97d3aa0 100644
--- a/tests/test_mcp.py
+++ b/tests/test_mcp.py
@@ -154,6 +154,98 @@ def test_skipped_when_binary_not_installed(self):
         assert "cursor" not in clients
 
 
+class TestKiloMcpClient:
+    def test_kilo_registered_in_mcp_clients(self):
+        assert "kilo" in mcp.MCP_CLIENTS
+        assert mcp.MCP_CLIENTS["kilo"]["binary"] == "kilo"
+        assert mcp.MCP_CLIENTS["kilo"]["list_command"] == "kilo mcp list"
+
+    def test_kilo_in_client_modules(self):
+        assert mcp._MCP_CLIENT_MODULES["kilo"] is mcp.kilo
+
+    def test_kilo_cli_env_pins_xdg_config_home(self):
+        # `kilo mcp list` must read ug's isolated config dir, so the env ug runs it
+        # with pins XDG_CONFIG_HOME to the same dir kilo.build_runtime_env uses.
+        env = mcp._kilo_cli_env()
+        assert env["XDG_CONFIG_HOME"] == str(mcp.kilo.KILO_XDG_CONFIG_HOME)
+
+    def test_parses_kilo_tree_mcp_list_output(self):
+        # Kilo prints `●  <glyph> <name> <status>` rows with indented `│` detail lines
+        # (the proxy command, which must NOT be parsed as a server).
+        output = (
+            "┌  MCP Servers\n"
+            "│\n"
+            "●  ✗ system-ai-atlassian failed\n"
+            "│      Failed to get tools\n"
+            "│      /path/ug mcp-proxy --url https://ws/ai-gateway/.../system.ai.atlassian\n"
+            "│\n"
+            "●  ✓ system-ai-dbsql connected\n"
+            "│      /path/ug mcp-proxy --url https://ws/ai-gateway/.../system.ai.dbsql\n"
+            "└  2 server(s)\n"
+        )
+        assert mcp.parse_mcp_list_output("kilo", output) == {
+            "system-ai-atlassian": mcp.LIVE_FAILED,
+            "system-ai-dbsql": mcp.LIVE_CONNECTED,
+        }
+
+    def test_kilo_empty_listing_parses_to_no_servers(self):
+        assert mcp.parse_mcp_list_output("kilo", "▲  No MCP servers configured") == {}
+
+    def test_run_mcp_list_passes_xdg_env_for_kilo(self, monkeypatch):
+        # Regression: without the pinned XDG_CONFIG_HOME, `kilo mcp list` reads the
+        # user's default ~/.config/kilo and reports ug's servers as missing.
+        captured: dict = {}
+
+        class _Result:
+            stdout = ""
+            stderr = ""
+
+        def _fake_run(argv, **kwargs):
+            captured["argv"] = argv
+            captured["env"] = kwargs.get("env")
+            return _Result()
+
+        monkeypatch.setattr(mcp.subprocess_cross_os, "run", _fake_run)
+        mcp._run_mcp_list("kilo")
+
+        assert captured["argv"] == ["kilo", "mcp", "list"]
+        assert captured["env"] is not None
+        assert captured["env"]["XDG_CONFIG_HOME"] == str(mcp.kilo.KILO_XDG_CONFIG_HOME)
+
+    def test_configure_dispatches_proxy_argv_to_kilo_writer(self, monkeypatch):
+        calls: list[tuple[str, list[str]]] = []
+        monkeypatch.setattr(
+            mcp.kilo,
+            "write_mcp_server_config",
+            lambda name, argv: calls.append((name, argv)) or False,
+        )
+
+        removed_scopes = mcp.configure_client_mcp_server("kilo", "github", GH_URL, WS, "p")
+
+        assert removed_scopes == []
+        assert calls == [("github", _proxy_argv())]
+
+    def test_configure_reports_user_scope_on_replace(self, monkeypatch):
+        monkeypatch.setattr(mcp.kilo, "write_mcp_server_config", lambda name, argv: True)
+        assert mcp.configure_client_mcp_server("kilo", "github", GH_URL, WS, "p") == [
+            mcp.MCP_USER_SCOPE
+        ]
+
+    def test_remove_dispatches_to_kilo_remover(self, monkeypatch):
+        calls: list[str] = []
+        monkeypatch.setattr(
+            mcp.kilo, "remove_mcp_server_config", lambda name: calls.append(name) or True
+        )
+        assert mcp.remove_client_mcp_server("kilo", "github-mcp") == [mcp.MCP_USER_SCOPE]
+        assert calls == ["github-mcp"]
+
+    def test_managed_entry_builds_via_kilo_module(self):
+        argv = mcp.build_mcp_proxy_argv(GH_URL, WS, None, use_pat=False)
+        assert mcp._managed_mcp_entry(
+            "kilo", GH_URL, WS, None, use_pat=False, always_load=False, http_client=None
+        ) == mcp.kilo.build_mcp_server_entry(argv)
+
+
 class TestConfigureClientMcpServer:
     def test_configures_copilot_with_proxy_argv(self, monkeypatch):
         calls: list[tuple[str, list[str]]] = []