feat(kimi): Kimi K3 provider on the official kimi-code CLI (#713)

* feat(kimi): Kimi K3 provider on the official kimi-code CLI (Wave 1)

ModelProvider.KIMI routes through KimiCliProvider driving Moonshot's kimi
CLI on a Kimi subscription (OAuth device-code, no metered key). One-shot
delivery roles only (V1), interactive ban wired in both guard lists.

Auth: one shared RW auth mount; containers symlink credentials/ and
oauth/ (the CLI's cross-process refresh-lock dir) into a container-local
KIMI_CODE_HOME so every container and the host redeem the SAME rotating
refresh chain - live-verified that per-copy chains cross-invalidate after
the reuse-grace window. No orchestrator refresh daemon; an expires_at
preflight exits 78.

Config renderer mirrors the login-managed provider/model blocks
field-for-field (live-captured; the model value is the CLI-side name,
never the raw API id), plus per-role deny rules and the bash-guard as a
PreToolUse hook via a wrapper script (an env key on a hooks entry makes
the CLI silently drop ALL hooks - live-verified). Usage capture sums
wire.jsonl usage.record 4-bucket events; sniff classifies rate-limit/auth
from structured error text only, mapped to the shared 75/78 park
contract. Image installs the CLI latest-at-build (no version pin, by
policy) with the resolved version stamped as provenance, binary split to
/usr/local away from mutable state.

Migrations 090 (enum) + 091 (provider seed); catalog, pricing, routing
mode, and orchestrator park/usage wiring mirror the codex integration.

* feat(kimi): surface sweep + fleet-wide pin drop (Wave 2)

Compose x3 gain the agent-kimi-image service and the orchestrator's
read-write ~/.kimi-code mount + kimi-usage dir; .env.example documents
the Kimi block. Panel mirrors ModelProvider.KIMI and adds the kimi
routing mode (catalog filter, mode button, mix-picker group, badge) with
tests; provider routes gain the kimi remediation entry. CLAUDE.md and
docs/map document the runtime. Per the no-pins policy, agent-grok/
gemini/codex Dockerfiles drop their version pins for latest-at-build
with resolved-version provenance stamps (grok resolves 0.2.112 vs the
old 0.2.56 pin - verified by real builds of all four images).

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

tests/integration/test_llm_routing.py

@@ -67,7 +67,15 @@ async def llm_setup(
         type=ModelProvider.GEMINI,
         enabled=True,
     )
-    db_session.add_all([anthropic, grok, ollama, openai, gemini])
+    # Mirrors migration 091_seed_kimi_provider's contract: enabled=True at
+    # seed time (Codex's shape, not Gemini's disabled-then-flipped one) — no
+    # base_url, subscription auth only.
+    kimi = ProviderConfigTable(
+        name="kimi-test",
+        type=ModelProvider.KIMI,
+        enabled=True,
+    )
+    db_session.add_all([anthropic, grok, ollama, openai, gemini, kimi])
     await db_session.flush()
     yield {"svc": ModelRoutingService(db_session)}
 
@@ -235,6 +243,18 @@ async def test_derive_mode_gemini_when_only_gemini_global(llm_setup: dict) -> No
     assert await svc.derive_mode() == "gemini"
 
 
+@pytest.mark.asyncio
+async def test_derive_mode_kimi_when_only_kimi_global(llm_setup: dict) -> None:
+    """A pure-KIMI global assignment reports "kimi", not the catch-all
+    "mix" — mirrors the codex/gemini branches derive_mode already carries."""
+    svc = llm_setup["svc"]
+    kimi_model = _first_model_for_type(ModelProvider.KIMI)
+    await svc.upsert_assignment(
+        scope=AssignmentScope.GLOBAL, scope_value=None, model_name=kimi_model
+    )
+    assert await svc.derive_mode() == "kimi"
+
+
 @pytest.mark.asyncio
 async def test_derive_mode_mix_with_per_agent(llm_setup: dict) -> None:
     svc = llm_setup["svc"]
@@ -419,6 +439,40 @@ async def test_apply_mode_gemini_enables_gemini_provider(llm_setup: dict) -> Non
     assert refetched.enabled is True
 
 
+@pytest.mark.asyncio
+async def test_apply_mode_kimi_sets_global(llm_setup: dict) -> None:
+    svc = llm_setup["svc"]
+    await svc.apply_mode(mode="kimi")
+    assignments = await svc.list_assignments()
+    assert len(assignments) == 1
+    assert assignments[0].scope == AssignmentScope.GLOBAL
+    assert assignments[0].provider.type == ModelProvider.KIMI
+    assert assignments[0].model_name == "kimi-code/k3"
+
+
+@pytest.mark.asyncio
+async def test_apply_mode_kimi_enables_kimi_provider(llm_setup: dict) -> None:
+    """apply_mode('kimi') force-enables the KIMI row — belt-and-suspenders
+    against a row disabled by some other path, mirroring the codex/gemini tests."""
+    svc = llm_setup["svc"]
+    provider_svc = ProviderService(svc.session)
+    kimi = next(
+        p
+        for p in await provider_svc.list_providers(include_disabled=True)
+        if p.type == ModelProvider.KIMI
+    )
+    await provider_svc.update_provider(
+        cast("UUID", kimi.id), ProviderUpdate(enabled=False)
+    )
+    await svc.session.flush()
+
+    await svc.apply_mode(mode="kimi")
+
+    refetched = await provider_svc.get_provider(cast("UUID", kimi.id))
+    assert refetched is not None
+    assert refetched.enabled is True
+
+
 @pytest.mark.asyncio
 async def test_apply_mode_mix_requires_per_agent(llm_setup: dict) -> None:
     svc = llm_setup["svc"]
@@ -604,6 +658,78 @@ async def test_upsert_assignment_enables_disabled_gemini_provider(
     assert route.provider_type == ModelProvider.GEMINI
 
 
+@pytest.mark.asyncio
+async def test_upsert_and_resolve_kimi_assignment_roundtrip(
+    llm_setup: dict,
+) -> None:
+    """kimi-code/k3 through upsert_assignment -> resolve_for_agent, against
+    the seeded KIMI row (migration 091). Proves resolve_for_agent actually
+    returns a KIMI spawn route — not a silent Anthropic fallback."""
+    svc = llm_setup["svc"]
+    kimi_model = _first_model_for_type(ModelProvider.KIMI)
+    row = await svc.upsert_assignment(
+        scope=AssignmentScope.AGENT_SLUG,
+        scope_value="be-dev-1",
+        model_name=kimi_model,
+    )
+    assert row.model_name == kimi_model
+
+    route = await svc.resolve_for_agent("be-dev-1")
+    assert route.provider_type == ModelProvider.KIMI
+    assert route.model_name == kimi_model
+    # The seeded row carries no stored token — Kimi authenticates via the
+    # shared, symlinked-in ~/.kimi-code subscription credential, not a
+    # decrypted token.
+    assert route.auth_token is None
+
+
+@pytest.mark.asyncio
+async def test_upsert_assignment_enables_disabled_kimi_provider(
+    llm_setup: dict,
+) -> None:
+    """Belt-and-suspenders: assigning a Kimi model via Mix (upsert_assignment)
+    force-enables the row even if it was disabled — not just apply_mode('kimi')."""
+    svc = llm_setup["svc"]
+    provider_svc = ProviderService(svc.session)
+    kimi = next(
+        p
+        for p in await provider_svc.list_providers(include_disabled=True)
+        if p.type == ModelProvider.KIMI
+    )
+    await provider_svc.update_provider(
+        cast("UUID", kimi.id), ProviderUpdate(enabled=False)
+    )
+    await svc.session.flush()
+
+    kimi_model = _first_model_for_type(ModelProvider.KIMI)
+    await svc.upsert_assignment(
+        scope=AssignmentScope.AGENT_SLUG,
+        scope_value="be-dev-1",
+        model_name=kimi_model,
+    )
+
+    refetched = await provider_svc.get_provider(cast("UUID", kimi.id))
+    assert refetched is not None
+    assert refetched.enabled is True
+    # And the route actually resolves to KIMI now that it's enabled.
+    route = await svc.resolve_for_agent("be-dev-1")
+    assert route.provider_type == ModelProvider.KIMI
+
+
+@pytest.mark.asyncio
+async def test_apply_mode_kimi_end_to_end_reachable(llm_setup: dict) -> None:
+    """The full reachability chain: apply_mode -> derive_mode reflects it ->
+    resolve_for_agent actually spawns Kimi, mirroring the Codex/Gemini tests."""
+    svc = llm_setup["svc"]
+    await svc.apply_mode(mode="kimi")
+
+    assert await svc.derive_mode() == "kimi"
+
+    route = await svc.resolve_for_agent("be-dev-1")
+    assert route.provider_type == ModelProvider.KIMI
+    assert route.model_name == "kimi-code/k3"
+
+
 @pytest.mark.asyncio
 async def test_apply_mode_gemini_end_to_end_reachable(llm_setup: dict) -> None:
     """The full reachability chain the original drill missed: apply_mode
@@ -632,12 +758,12 @@ async def test_apply_mode_codex_end_to_end_reachable(llm_setup: dict) -> None:
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("mode", ["codex", "gemini"])
+@pytest.mark.parametrize("mode", ["codex", "gemini", "kimi"])
 @pytest.mark.parametrize("interactive_slug", ["intake-1", "secretary-1"])
 async def test_interactive_agents_exempt_from_delivery_only_global_mode(
     llm_setup: dict, mode: str, interactive_slug: str
 ) -> None:
-    """A fleet-wide Codex/Gemini mode must not capture Intake/Secretary —
+    """A fleet-wide Codex/Gemini/Kimi mode must not capture Intake/Secretary —
     they have no V1 support on those providers, so the resolver keeps them
     on the legacy Anthropic path (the completeness-drill gap: previously
     they resolved to the unsupported provider and the spawn guard left both

tests/integration/test_migration_091_seed_kimi_provider.py

@@ -0,0 +1,162 @@
+"""Migration 091 tests — seed_kimi_provider.
+
+Verifies the post-upgrade state and exercises the downgrade SQL ordering,
+mirroring ``test_migration_083_seed_openai_provider.py``'s own shape (Kimi's
+row is seeded ``enabled=true`` directly, the same posture as Codex's — there
+is no ``apply_mode="kimi"``-only gate it needs to wait behind, unlike GROK's
+disabled-until-key-set seed).
+
+NOT a real alembic round-trip — the suite builds the test DB via
+Base.metadata.create_all (see conftest). Migration 091's upgrade()/downgrade()
+bodies are reviewed here; the tests guard the resulting DB-level contract —
+in particular ``enabled=True`` at seed time and NULL base_url/auth_token
+(subscription auth, no stored secret).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from uuid import uuid4
+
+import pytest
+from roboco.db.tables import ModelAssignmentTable, ProviderConfigTable
+from roboco.models.base import AssignmentScope, ModelProvider
+from sqlalchemy import text
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+_INSERT_SQL = text(
+    """
+    INSERT INTO provider_configs
+        (id, name, type, base_url, auth_token_encrypted, enabled, created_at)
+    VALUES
+        (
+            gen_random_uuid(),
+            'Kimi (Moonshot)',
+            'kimi',
+            NULL,
+            NULL,
+            true,
+            now()
+        )
+    ON CONFLICT (name) DO NOTHING
+    """
+)
+
+
+@pytest.mark.asyncio
+async def test_migration_091_upgrade_insert_contract(
+    db_session: AsyncSession,
+) -> None:
+    """The upgrade INSERT SQL seeds the Kimi row ENABLED with no stored
+    secret (subscription auth), and is idempotent."""
+    # --- First run: the row should be inserted.
+    await db_session.execute(_INSERT_SQL)
+    await db_session.flush()
+
+    result = await db_session.execute(
+        text(
+            "SELECT name, type, enabled, base_url, auth_token_encrypted "
+            "FROM provider_configs "
+            "WHERE name = 'Kimi (Moonshot)'"
+        )
+    )
+    rows = list(result)
+    assert len(rows) == 1
+    name, ptype, enabled, base_url, auth_token = rows[0]
+    assert name == "Kimi (Moonshot)"
+    assert ptype == "kimi"
+    # The load-bearing assertion: enabled=True at seed time (parity with
+    # Codex — there's no key-collection step gating this row).
+    assert enabled is True
+    assert base_url is None
+    assert auth_token is None
+
+    # --- Second run: ON CONFLICT DO NOTHING must not create a duplicate.
+    await db_session.execute(_INSERT_SQL)
+    await db_session.flush()
+
+    result = await db_session.execute(
+        text("SELECT id FROM provider_configs WHERE name = 'Kimi (Moonshot)'")
+    )
+    assert len(list(result)) == 1, (
+        "Expected exactly one 'Kimi (Moonshot)' row after two INSERT "
+        "executions; ON CONFLICT DO NOTHING must prevent duplicates."
+    )
+
+
+@pytest.mark.asyncio
+async def test_migration_091_downgrade_deletes_assignments_before_config(
+    db_session: AsyncSession,
+) -> None:
+    """Downgrade SQL deletes model_assignments before provider_configs.
+
+    A FK RESTRICT constraint on model_assignments.provider_config_id means
+    deleting provider_configs first would raise an IntegrityError.
+    """
+    suffix = uuid4().hex[:8]
+    kimi = ProviderConfigTable(
+        name=f"Kimi (Moonshot)-test-{suffix}",
+        type=ModelProvider.KIMI,
+        enabled=True,
+    )
+    db_session.add(kimi)
+    await db_session.flush()
+
+    assignment = ModelAssignmentTable(
+        scope=AssignmentScope.AGENT_SLUG,
+        scope_value=f"test-agent-{suffix}",
+        provider_config_id=kimi.id,
+        model_name="kimi-code/k3",
+    )
+    db_session.add(assignment)
+    await db_session.flush()
+
+    result = await db_session.execute(
+        text("SELECT id FROM provider_configs WHERE name = :name").bindparams(
+            name=kimi.name
+        )
+    )
+    assert result.scalar_one_or_none() is not None
+
+    result = await db_session.execute(
+        text("SELECT id FROM model_assignments WHERE scope_value = :sv").bindparams(
+            sv=assignment.scope_value
+        )
+    )
+    assert result.scalar_one_or_none() is not None
+
+    # Step 1: delete referencing model_assignments first.
+    await db_session.execute(
+        text(
+            "DELETE FROM model_assignments "
+            "WHERE provider_config_id IN ("
+            "    SELECT id FROM provider_configs WHERE name = :name"
+            ")"
+        ).bindparams(name=kimi.name)
+    )
+    # Step 2: now safe to delete the provider row.
+    await db_session.execute(
+        text("DELETE FROM provider_configs WHERE name = :name").bindparams(
+            name=kimi.name
+        )
+    )
+
+    result = await db_session.execute(
+        text("SELECT id FROM provider_configs WHERE name = :name").bindparams(
+            name=kimi.name
+        )
+    )
+    assert result.scalar_one_or_none() is None, (
+        "provider_configs row should be deleted by downgrade"
+    )
+
+    result = await db_session.execute(
+        text("SELECT id FROM model_assignments WHERE scope_value = :sv").bindparams(
+            sv=assignment.scope_value
+        )
+    )
+    assert result.scalar_one_or_none() is None, (
+        "model_assignments row should be deleted before provider_configs"
+    )

tests/unit/billing/test_pricing.py

@@ -77,6 +77,23 @@ _GLM_OUTPUT = 4.40
 _GLM_CACHE_READ = 0.26
 _GLM_CACHE_WRITE = 1.40
 
+# Moonshot Kimi — priced non-Anthropic (kimi-code CLI subscription, priced
+# here for cost attribution like grok-build/gpt-5.3-codex).
+_KIMI_K3_INPUT = 3.00
+_KIMI_K3_OUTPUT = 15.00
+_KIMI_K3_CACHE_READ = 0.30
+_KIMI_K3_CACHE_WRITE = 3.00
+
+_KIMI_CODING_INPUT = 0.95
+_KIMI_CODING_OUTPUT = 4.00
+_KIMI_CODING_CACHE_READ = 0.19
+_KIMI_CODING_CACHE_WRITE = 0.95
+
+_KIMI_CODING_HIGHSPEED_INPUT = 1.90
+_KIMI_CODING_HIGHSPEED_OUTPUT = 8.00
+_KIMI_CODING_HIGHSPEED_CACHE_READ = 0.38
+_KIMI_CODING_HIGHSPEED_CACHE_WRITE = 1.90
+
 # Tolerance for floating-point comparisons
 _TOL = 1e-4
 
@@ -395,6 +412,78 @@ class TestCodexTier:
         assert _CODEX_OUTPUT > _CODEX_INPUT
 
 
+# ---------------------------------------------------------------------------
+# Kimi tier (Moonshot — priced non-Anthropic, four login-managed aliases)
+# ---------------------------------------------------------------------------
+
+
+class TestKimiTier:
+    """kimi-code/* pricing — cache_write folds to the input rate, same
+    convention as grok-build/gpt-5.3-codex (no published cache-write discount)."""
+
+    def test_k3_all_token_types(self) -> None:
+        cost = calculate_cost(
+            "kimi-code/k3",
+            tokens_input=_M,
+            tokens_output=_M,
+            tokens_cache_read=_M,
+            tokens_cache_write=_M,
+        )
+        expected = (
+            _KIMI_K3_INPUT
+            + _KIMI_K3_OUTPUT
+            + _KIMI_K3_CACHE_READ
+            + _KIMI_K3_CACHE_WRITE
+        )
+        assert abs(cost - expected) < _TOL
+
+    def test_k3_256k_prices_the_same_as_k3(self) -> None:
+        # "kimi-code/k3" is a PREFIX of "kimi-code/k3-256k" — longest-fragment
+        # wins in _lookup_prices must resolve the 256k alias to its own entry,
+        # not silently fall through to the bare k3 fragment (same rates here,
+        # but the resolution path is what's under test).
+        cost = calculate_cost("kimi-code/k3-256k", tokens_input=_M, tokens_output=0)
+        assert abs(cost - _KIMI_K3_INPUT) < _TOL
+
+    def test_kimi_for_coding_all_token_types(self) -> None:
+        cost = calculate_cost(
+            "kimi-code/kimi-for-coding",
+            tokens_input=_M,
+            tokens_output=_M,
+            tokens_cache_read=_M,
+            tokens_cache_write=_M,
+        )
+        expected = (
+            _KIMI_CODING_INPUT
+            + _KIMI_CODING_OUTPUT
+            + _KIMI_CODING_CACHE_READ
+            + _KIMI_CODING_CACHE_WRITE
+        )
+        assert abs(cost - expected) < _TOL
+
+    def test_kimi_for_coding_highspeed_resolves_its_own_longer_fragment(self) -> None:
+        # "kimi-code/kimi-for-coding" is a PREFIX of
+        # "kimi-code/kimi-for-coding-highspeed" — longest-fragment-wins must
+        # resolve the highspeed alias to its OWN (pricier) rate, not the base
+        # coding tier's cheaper one.
+        cost = calculate_cost(
+            "kimi-code/kimi-for-coding-highspeed", tokens_input=_M, tokens_output=0
+        )
+        assert abs(cost - _KIMI_CODING_HIGHSPEED_INPUT) < _TOL
+        assert cost > calculate_cost(
+            "kimi-code/kimi-for-coding", tokens_input=_M, tokens_output=0
+        )
+
+    def test_kimi_is_not_treated_as_anthropic(self) -> None:
+        assert _is_anthropic_model("kimi-code/k3") is False
+        assert calculate_cost("kimi-code/k3", tokens_input=_M, tokens_output=0) > 0.0
+
+    def test_output_is_pricier_than_input(self) -> None:
+        assert _KIMI_K3_OUTPUT > _KIMI_K3_INPUT
+        assert _KIMI_CODING_OUTPUT > _KIMI_CODING_INPUT
+        assert _KIMI_CODING_HIGHSPEED_OUTPUT > _KIMI_CODING_HIGHSPEED_INPUT
+
+
 # ---------------------------------------------------------------------------
 # GLM-5.2 tier (Ollama Cloud — priced non-Anthropic, grounded in a citable
 # published rate; see the module's pricing-table comment for the source).
@@ -620,6 +709,12 @@ class TestCostResult:
         assert result.unpriced is False
         assert result.is_anthropic is False
 
+    def test_priced_non_anthropic_kimi_is_not_unpriced(self) -> None:
+        result = calculate_cost_result("kimi-code/k3", tokens_input=_M, tokens_output=0)
+        assert result.cost_usd > 0.0
+        assert result.unpriced is False
+        assert result.is_anthropic is False
+
     def test_calculate_cost_matches_structured_cost_usd(self) -> None:
         model = "claude-opus-5"
         assert (

tests/unit/llm/providers/test_kimi_cli_config.py

@@ -0,0 +1,333 @@
+"""kimi_cli_config — config.toml (managed blocks + per-role permission rules
++ hooks) + mcp.json passthrough + AGENTS.md + the auth preflight."""
+
+from __future__ import annotations
+
+import json
+import tomllib
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING
+
+from roboco.llm.providers import kimi_cli_config as kc
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    import pytest
+
+_SAMPLE_MCP = {
+    "mcpServers": {
+        "roboco-flow": {
+            "command": "uv",
+            "args": ["run", "--no-sync", "python", "-m", "roboco.mcp.flow_server"],
+            "env": {"ROBOCO_AGENT_ID": "be-dev-1", "ROBOCO_AGENT_TOKEN": "tok-123"},
+        },
+        "roboco-do": {"command": "uv", "args": ["run", "x"]},
+    }
+}
+
+
+# ---------------------------------------------------------------------------
+# permission_rules_for_role — deny-only, role-scoped
+# ---------------------------------------------------------------------------
+
+
+def test_fleet_wide_denies_present_for_every_role() -> None:
+    for role in ("developer", "qa", "pr_reviewer", "main_pm", "unknown-role-xyz"):
+        rules = kc.permission_rules_for_role(role)
+        patterns = {r["pattern"] for r in rules}
+        for fleet_wide in kc._FLEET_WIDE_DENY:
+            assert fleet_wide in patterns
+        assert all(r["decision"] == "deny" for r in rules)
+
+
+def test_bash_capable_role_keeps_bash_and_denies_git_destructive_pm() -> None:
+    rules = kc.permission_rules_for_role("developer")
+    patterns = {r["pattern"] for r in rules}
+    assert "Bash" not in patterns  # bash-capable: no blanket deny
+    assert "Bash(git push*)" in patterns
+    assert "Bash(rm -rf*)" in patterns
+    assert "Bash(uv run*)" in patterns
+    # Developer writes code — no edit-tool deny.
+    assert "Write" not in patterns
+    assert "Edit" not in patterns
+
+
+def test_non_bash_role_gets_blanket_bash_deny_and_no_command_scoped_rules() -> None:
+    rules = kc.permission_rules_for_role("pr_reviewer")
+    patterns = {r["pattern"] for r in rules}
+    assert "Bash" in patterns
+    assert "Bash(git push*)" not in patterns  # blanket deny — nothing left to scope
+    # Read-only reviewer doesn't write code either.
+    assert "Write" in patterns
+    assert "Edit" in patterns
+
+
+def test_main_pm_keeps_bash_but_denies_write_edit() -> None:
+    rules = kc.permission_rules_for_role("main_pm")
+    patterns = {r["pattern"] for r in rules}
+    assert "Bash" not in patterns  # PM keeps its shell
+    assert "Bash(git push*)" in patterns
+    assert "Write" in patterns  # PM doesn't write code
+    assert "Edit" in patterns
+
+
+def test_unknown_role_gets_every_deny_category() -> None:
+    rules = kc.permission_rules_for_role("unknown-role-xyz")
+    patterns = {r["pattern"] for r in rules}
+    assert "Write" in patterns
+    assert "Edit" in patterns
+    assert "Bash" in patterns
+
+
+# ---------------------------------------------------------------------------
+# kimi_hooks_config
+# ---------------------------------------------------------------------------
+
+
+def test_kimi_hooks_config_wires_bash_guard_wrapper_no_env_field() -> None:
+    # A [[hooks]] entry with an `env` key gets the WHOLE hooks section
+    # silently dropped by the CLI (live-verified) — env delivery must ride
+    # the wrapper script's own export, never a rendered `env` field.
+    hooks = kc.kimi_hooks_config("/app/scripts/kimi-bash-guard-wrapper.sh")
+    assert len(hooks) == 1
+    hook = hooks[0]
+    assert hook["event"] == "PreToolUse"
+    assert hook["matcher"] == "Bash"
+    assert hook["command"] == "/app/scripts/kimi-bash-guard-wrapper.sh"
+    assert "env" not in hook
+
+
+def test_kimi_hooks_config_default_points_at_wrapper() -> None:
+    hooks = kc.kimi_hooks_config()
+    assert hooks[0]["command"] == kc.KIMI_BASH_GUARD_WRAPPER
+    assert hooks[0]["command"].endswith("kimi-bash-guard-wrapper.sh")
+
+
+def test_kimi_hooks_config_entries_only_carry_legal_keys() -> None:
+    # Pins the whole defect class: any future field addition to a rendered
+    # hook entry that isn't one of these four gets silently dropped by kimi.
+    legal_keys = {"event", "matcher", "command", "timeout"}
+    for hook in kc.kimi_hooks_config():
+        assert set(hook.keys()) <= legal_keys
+
+
+# ---------------------------------------------------------------------------
+# render_config_toml — valid TOML, managed blocks + telemetry/upgrade + rules
+# ---------------------------------------------------------------------------
+
+
+def test_render_config_toml_is_valid_toml() -> None:
+    parsed = tomllib.loads(kc.render_config_toml("developer"))
+    assert parsed["telemetry"] is False
+    assert parsed["upgrade"]["auto_install"] is False
+
+
+def test_render_config_toml_managed_provider_block() -> None:
+    parsed = tomllib.loads(kc.render_config_toml("developer"))
+    provider = parsed["providers"]["managed:kimi-code"]
+    assert provider["type"] == "kimi"
+    assert provider["base_url"] == "https://api.kimi.com/coding/v1"
+    assert provider["oauth"]["storage"] == "file"
+    assert provider["oauth"]["key"] == "oauth/kimi-code"
+
+
+def test_render_config_toml_carries_all_four_model_aliases() -> None:
+    parsed = tomllib.loads(kc.render_config_toml("developer"))
+    models = parsed["models"]
+    for alias in (
+        "kimi-code/k3",
+        "kimi-code/k3-256k",
+        "kimi-code/kimi-for-coding",
+        "kimi-code/kimi-for-coding-highspeed",
+    ):
+        assert alias in models
+        assert models[alias]["provider"] == "managed:kimi-code"
+        assert models[alias]["max_context_size"] > 0
+        # The `model` value is the CLI-side managed name the wire sees —
+        # exactly the alias's last segment, never a raw API id like
+        # "kimi-k3" (a live-capture drift that would break every run).
+        assert models[alias]["model"] == alias.removeprefix("kimi-code/")
+        assert "thinking" in models[alias]["capabilities"]
+    # Only the K3 family exposes reasoning effort knobs.
+    assert models["kimi-code/k3"]["default_effort"] == "high"
+    assert "default_effort" not in models["kimi-code/kimi-for-coding"]
+
+
+def test_render_config_toml_services_share_the_managed_oauth() -> None:
+    parsed = tomllib.loads(kc.render_config_toml("developer"))
+    for service in ("moonshot_search", "moonshot_fetch"):
+        assert parsed["services"][service]["oauth"]["key"] == "oauth/kimi-code"
+
+
+def test_render_config_toml_permission_rules_vary_by_role() -> None:
+    # developer keeps its shell -> gets the full command-scoped git/destructive/
+    # raw-PM deny list underneath it; pr_reviewer's blanket Bash deny needs no
+    # command-scoped rules at all, so it ends up with FEWER total rules despite
+    # also denying Write/Edit on top of the fleet-wide set.
+    dev_rules = tomllib.loads(kc.render_config_toml("developer"))["permission"]["rules"]
+    reviewer_rules = tomllib.loads(kc.render_config_toml("pr_reviewer"))["permission"][
+        "rules"
+    ]
+    assert len(dev_rules) > len(reviewer_rules)
+
+
+def test_render_config_toml_hooks_present() -> None:
+    parsed = tomllib.loads(kc.render_config_toml("developer"))
+    assert parsed["hooks"][0]["event"] == "PreToolUse"
+
+
+# ---------------------------------------------------------------------------
+# render_mcp_json — near-passthrough of the mounted mcp-config.json
+# ---------------------------------------------------------------------------
+
+
+def test_render_mcp_json_injects_env_and_omits_empty_env() -> None:
+    rendered = json.loads(kc.render_mcp_json(_SAMPLE_MCP))
+    flow = rendered["mcpServers"]["roboco-flow"]
+    assert flow["command"] == "uv"
+    assert flow["args"][:2] == ["run", "--no-sync"]
+    assert flow["env"]["ROBOCO_AGENT_TOKEN"] == "tok-123"
+    assert "env" not in rendered["mcpServers"]["roboco-do"]
+
+
+def test_render_mcp_json_empty_servers() -> None:
+    assert json.loads(kc.render_mcp_json({})) == {"mcpServers": {}}
+
+
+# ---------------------------------------------------------------------------
+# write_agents_md
+# ---------------------------------------------------------------------------
+
+
+def test_write_agents_md_installs_the_blueprint(tmp_path: Path) -> None:
+    src = tmp_path / "system-prompt.md"
+    src.write_text("You are a RoboCo backend developer.", encoding="utf-8")
+    dest = tmp_path / ".kimi-code" / "AGENTS.md"
+    assert kc.write_agents_md(source=src, dest=dest) is True
+    assert dest.read_text(encoding="utf-8") == "You are a RoboCo backend developer."
+
+
+def test_write_agents_md_noops_when_source_absent(tmp_path: Path) -> None:
+    dest = tmp_path / ".kimi-code" / "AGENTS.md"
+    assert kc.write_agents_md(source=tmp_path / "absent.md", dest=dest) is False
+    assert not dest.exists()
+
+
+# ---------------------------------------------------------------------------
+# Auth preflight — a plain expires_at JSON field, no JWT decode
+# ---------------------------------------------------------------------------
+
+
+def _write_creds(path: Path, *, expires_at: object) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps(
+            {
+                "access_token": "at",
+                "refresh_token": "rt",
+                "expires_at": expires_at,
+                "expires_in": 900,
+                "scope": "chat",
+                "token_type": "Bearer",
+            }
+        ),
+        encoding="utf-8",
+    )
+
+
+def test_is_valid_true_for_future_unix_timestamp(tmp_path: Path) -> None:
+    creds = tmp_path / "credentials" / "kimi-code.json"
+    future = datetime.now(UTC) + timedelta(minutes=10)
+    _write_creds(creds, expires_at=future.timestamp())
+    assert kc.is_valid(creds) is True
+
+
+def test_is_valid_false_for_past_unix_timestamp(tmp_path: Path) -> None:
+    creds = tmp_path / "credentials" / "kimi-code.json"
+    past = datetime.now(UTC) - timedelta(minutes=10)
+    _write_creds(creds, expires_at=past.timestamp())
+    assert kc.is_valid(creds) is False
+
+
+def test_is_valid_accepts_iso8601_string(tmp_path: Path) -> None:
+    creds = tmp_path / "credentials" / "kimi-code.json"
+    future = datetime.now(UTC) + timedelta(minutes=10)
+    _write_creds(creds, expires_at=future.isoformat())
+    assert kc.is_valid(creds) is True
+
+
+def test_is_valid_false_for_missing_file(tmp_path: Path) -> None:
+    assert kc.is_valid(tmp_path / "credentials" / "kimi-code.json") is False
+
+
+def test_is_valid_false_for_unparseable_expires_at(tmp_path: Path) -> None:
+    creds = tmp_path / "credentials" / "kimi-code.json"
+    _write_creds(creds, expires_at="not-a-timestamp")
+    assert kc.is_valid(creds) is False
+
+
+def test_seconds_until_expiry_respects_skew(tmp_path: Path) -> None:
+    creds = tmp_path / "credentials" / "kimi-code.json"
+    soon = datetime.now(UTC) + timedelta(seconds=30)
+    _write_creds(creds, expires_at=soon.timestamp())
+    assert kc.is_valid(creds, skew_seconds=60) is False
+    assert kc.is_valid(creds, skew_seconds=0) is True
+
+
+# ---------------------------------------------------------------------------
+# main() — render mode + --check mode
+# ---------------------------------------------------------------------------
+
+
+def test_main_writes_config_mcp_and_agents_md(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    mcp_path = tmp_path / "mcp-config.json"
+    mcp_path.write_text(json.dumps(_SAMPLE_MCP), encoding="utf-8")
+    config_path = tmp_path / ".kimi-code" / "config.toml"
+    mcp_out_path = tmp_path / ".kimi-code" / "mcp.json"
+    agents_md_path = tmp_path / ".kimi-code" / "AGENTS.md"
+    system_prompt = tmp_path / "system-prompt.md"
+    system_prompt.write_text("blueprint", encoding="utf-8")
+
+    monkeypatch.setattr(kc, "KIMI_CONFIG_PATH", config_path)
+    monkeypatch.setattr(kc, "KIMI_MCP_PATH", mcp_out_path)
+    monkeypatch.setattr(kc, "KIMI_AGENTS_MD_PATH", agents_md_path)
+    monkeypatch.setattr(kc, "SYSTEM_PROMPT_PATH", system_prompt)
+    monkeypatch.setenv("ROBOCO_AGENT_ID", "be-dev-1")
+    monkeypatch.setenv("ROBOCO_MCP_CONFIG", str(mcp_path))
+
+    assert kc.main([]) == 0
+
+    parsed = tomllib.loads(config_path.read_text(encoding="utf-8"))
+    assert parsed["providers"]["managed:kimi-code"]["type"] == "kimi"
+    rendered_mcp = json.loads(mcp_out_path.read_text(encoding="utf-8"))
+    assert rendered_mcp["mcpServers"]["roboco-flow"]["env"]["ROBOCO_AGENT_TOKEN"] == (
+        "tok-123"
+    )
+    assert agents_md_path.read_text(encoding="utf-8") == "blueprint"
+
+
+def test_main_check_flag_runs_preflight_without_rendering(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    creds = tmp_path / "credentials" / "kimi-code.json"
+    future = datetime.now(UTC) + timedelta(minutes=10)
+    _write_creds(creds, expires_at=future.timestamp())
+    config_path = tmp_path / ".kimi-code" / "config.toml"
+
+    monkeypatch.setattr(kc, "KIMI_CREDENTIALS_PATH", creds)
+    monkeypatch.setattr(kc, "KIMI_CONFIG_PATH", config_path)
+
+    assert kc.main(["--check"]) == 0
+    assert not config_path.exists()  # --check never renders
+
+
+def test_main_check_flag_fails_on_missing_credential(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(
+        kc, "KIMI_CREDENTIALS_PATH", tmp_path / "credentials" / "kimi-code.json"
+    )
+    assert kc.main(["--check"]) == 1

tests/unit/llm/providers/test_kimi_cli_sniff.py

@@ -0,0 +1,186 @@
+"""kimi_cli_sniff — classify a Kimi run from ONLY its machine-relevant text.
+
+The structural guarantee under test: the model's own on-topic prose (which
+can legitimately contain the words "quota-limited" or a "429"/"401" substring
+inside a commit hash / id) must NEVER reach the classifier, because
+extraction only pulls a structured ``error`` field off error-bearing JSONL
+events plus raw stderr — never ``role: assistant`` / ``role: tool`` content.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING
+
+from roboco.llm.providers import kimi_cli_sniff as sniff
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    import pytest
+
+
+def _write_jsonl(path: Path, lines: list[str]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def _error_event(message: str) -> str:
+    return json.dumps({"type": "error", "error": {"message": message}})
+
+
+# ---------------------------------------------------------------------------
+# extract_error_text — structural isolation
+# ---------------------------------------------------------------------------
+
+
+def test_extract_error_text_pulls_only_structured_error_field(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(
+        log,
+        [
+            json.dumps(
+                {
+                    "role": "assistant",
+                    "content": "the quota-limited rollout ships this sprint",
+                }
+            ),
+            _error_event("real error text"),
+        ],
+    )
+    assert sniff.extract_error_text(log) == "real error text"
+
+
+def test_extract_error_text_accepts_bare_string_error() -> None:
+    assert sniff._error_text_from_event({"error": "bare string error"}) == (
+        "bare string error"
+    )
+
+
+def test_extract_error_text_empty_for_missing_or_error_less_log(
+    tmp_path: Path,
+) -> None:
+    assert sniff.extract_error_text(tmp_path / "nope.jsonl") == ""
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [json.dumps({"role": "assistant", "content": "hi"})])
+    assert sniff.extract_error_text(log) == ""
+
+
+# ---------------------------------------------------------------------------
+# The false-positive class this module exists to kill
+# ---------------------------------------------------------------------------
+
+
+def test_benign_transcript_never_false_parks(tmp_path: Path) -> None:
+    """A transcript whose ONLY content is benign on-topic prose — mentioning
+    "quota-limited" work and a commit hash containing "429"/"401" — must
+    classify as "" (no park), because none of it lives in a structured error
+    field the extractor even looks at."""
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(
+        log,
+        [
+            json.dumps(
+                {
+                    "role": "assistant",
+                    "content": (
+                        "Fixed the quota-limited rollout gate. Committed as "
+                        "abc4291f, also touched item 40199."
+                    ),
+                }
+            ),
+            json.dumps({"role": "tool", "tool_call_id": "1", "content": "ok"}),
+        ],
+    )
+    err_log = tmp_path / "run.err"
+    err_log.write_text("", encoding="utf-8")
+    assert sniff.classify(log, err_log) == ""
+
+
+def test_word_boundary_prevents_429_substring_false_positive() -> None:
+    assert not sniff.is_rate_limited("commit abc14293 deployed to prod")
+    assert not sniff.is_rate_limited("fix4297abc landed")
+
+
+def test_word_boundary_prevents_401_substring_false_positive() -> None:
+    assert not sniff.is_auth_failure("item 40199 was resolved")
+    assert not sniff.is_auth_failure("ticket 14012 closed")
+
+
+# ---------------------------------------------------------------------------
+# True positives — the live-verified error text shapes from the spike
+# ---------------------------------------------------------------------------
+
+
+def test_status_code_429_classifies_rate_limit(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_error_event("request failed with status code: 429")])
+    assert sniff.classify(log) == "rate_limit"
+
+
+def test_engine_overloaded_classifies_rate_limit(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_error_event("the engine is currently overloaded")])
+    assert sniff.classify(log) == "rate_limit"
+
+
+def test_usage_limit_for_period_classifies_rate_limit(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_error_event("usage limit for this period exceeded")])
+    assert sniff.classify(log) == "rate_limit"
+
+
+def test_usage_limit_for_billing_cycle_classifies_rate_limit(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_error_event("usage limit for this billing cycle reached")])
+    assert sniff.classify(log) == "rate_limit"
+
+
+def test_api_key_invalid_classifies_auth(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_error_event("API Key appears to be invalid")])
+    assert sniff.classify(log) == "auth"
+
+
+def test_membership_benefits_classifies_auth(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(
+        log,
+        [_error_event("We're unable to verify your membership benefits at this time.")],
+    )
+    assert sniff.classify(log) == "auth"
+
+
+def test_classify_reads_stderr_too(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [json.dumps({"role": "assistant", "content": "ok"})])
+    err_log = tmp_path / "run.err"
+    err_log.write_text("fatal: status code: 429\n", encoding="utf-8")
+    assert sniff.classify(log, err_log) == "rate_limit"
+
+
+def test_classify_missing_files_returns_empty(tmp_path: Path) -> None:
+    assert sniff.classify(tmp_path / "nope.jsonl", tmp_path / "nope.err") == ""
+
+
+def test_rate_limit_checked_before_auth_when_both_present(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(
+        log,
+        [_error_event("status code: 429, and API Key appears to be invalid too")],
+    )
+    assert sniff.classify(log) == "rate_limit"
+
+
+def test_main_cli_prints_classification(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_error_event("status code: 429")])
+    assert sniff.main([str(log)]) == 0
+    assert capsys.readouterr().out.strip() == "rate_limit"
+
+
+def test_main_cli_no_args_prints_empty(capsys: pytest.CaptureFixture[str]) -> None:
+    assert sniff.main([]) == 0
+    assert capsys.readouterr().out.strip() == ""

tests/unit/llm/providers/test_kimi_cli_usage.py

@@ -0,0 +1,264 @@
+"""kimi_cli_usage — resolve the session dir from a run's stdout meta line (or
+the newest matching session dir), then sum the real 4-bucket
+``usage.record``/``usageScope=="turn"`` events in that session's wire.jsonl.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from typing import TYPE_CHECKING
+
+from roboco.llm.providers import kimi_cli_usage as ku
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    import pytest
+
+
+def _write_jsonl(path: Path, lines: list[str]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def _usage_record(
+    *,
+    input_other: int,
+    output: int,
+    cache_read: int = 0,
+    cache_creation: int = 0,
+    scope: str = "turn",
+) -> str:
+    return json.dumps(
+        {
+            "type": "usage.record",
+            "model": "kimi-code/k3",
+            "usageScope": scope,
+            "usage": {
+                "inputOther": input_other,
+                "output": output,
+                "inputCacheRead": cache_read,
+                "inputCacheCreation": cache_creation,
+            },
+        }
+    )
+
+
+def _resume_hint(session_id: str) -> str:
+    return json.dumps(
+        {"role": "meta", "type": "session.resume_hint", "session_id": session_id}
+    )
+
+
+# ---------------------------------------------------------------------------
+# session_id_from_run_log
+# ---------------------------------------------------------------------------
+
+
+def test_session_id_from_run_log_finds_terminal_meta_line(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(
+        log,
+        [
+            json.dumps({"role": "assistant", "content": "working"}),
+            _resume_hint("session_abc123"),
+        ],
+    )
+    assert ku.session_id_from_run_log(log) == "session_abc123"
+
+
+def test_session_id_from_run_log_keeps_the_last_match(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [_resume_hint("session_first"), _resume_hint("session_second")])
+    assert ku.session_id_from_run_log(log) == "session_second"
+
+
+def test_session_id_from_run_log_none_when_absent(tmp_path: Path) -> None:
+    log = tmp_path / "run.jsonl"
+    _write_jsonl(log, [json.dumps({"role": "assistant", "content": "hi"})])
+    assert ku.session_id_from_run_log(log) is None
+    assert ku.session_id_from_run_log(tmp_path / "nope.jsonl") is None
+
+
+# ---------------------------------------------------------------------------
+# resolve_session_dir — primary (known id) + fallback (newest under cwd basename)
+# ---------------------------------------------------------------------------
+
+
+def test_resolve_session_dir_finds_known_session_id(tmp_path: Path) -> None:
+    home = tmp_path / ".kimi-code"
+    session_dir = home / "sessions" / "wd_myrepo_ab12cd34ef56" / "session_abc123"
+    session_dir.mkdir(parents=True)
+    resolved = ku.resolve_session_dir(
+        session_id="session_abc123",
+        workdir="/data/workspaces/myrepo",
+        kimi_code_home=home,
+    )
+    assert resolved == session_dir
+
+
+def test_resolve_session_dir_falls_back_to_newest(tmp_path: Path) -> None:
+    home = tmp_path / ".kimi-code"
+    wd_dir = home / "sessions" / "wd_myrepo_ab12cd34ef56"
+    old_session = wd_dir / "session_old"
+    new_session = wd_dir / "session_new"
+    old_session.mkdir(parents=True)
+    time.sleep(0.01)
+    new_session.mkdir(parents=True)
+    resolved = ku.resolve_session_dir(
+        session_id=None, workdir="/data/workspaces/myrepo", kimi_code_home=home
+    )
+    assert resolved == new_session
+
+
+def test_resolve_session_dir_none_when_sessions_root_absent(tmp_path: Path) -> None:
+    home = tmp_path / ".kimi-code"
+    assert (
+        ku.resolve_session_dir(
+            session_id=None, workdir="/x/myrepo", kimi_code_home=home
+        )
+        is None
+    )
+
+
+def test_resolve_session_dir_falls_back_when_id_not_found(tmp_path: Path) -> None:
+    home = tmp_path / ".kimi-code"
+    wd_dir = home / "sessions" / "wd_myrepo_ab12cd34ef56"
+    only_session = wd_dir / "session_other"
+    only_session.mkdir(parents=True)
+    resolved = ku.resolve_session_dir(
+        session_id="session_missing",
+        workdir="/data/workspaces/myrepo",
+        kimi_code_home=home,
+    )
+    assert resolved == only_session
+
+
+# ---------------------------------------------------------------------------
+# aggregate_usage_from_wire
+# ---------------------------------------------------------------------------
+
+
+def test_aggregate_sums_turn_scoped_usage_records(tmp_path: Path) -> None:
+    wire = tmp_path / "wire.jsonl"
+    _write_jsonl(
+        wire,
+        [
+            _usage_record(input_other=100, output=50, cache_read=10),
+            json.dumps({"type": "llm.request", "model": "kimi-code/k3"}),
+            _usage_record(input_other=200, output=80, cache_read=20, cache_creation=5),
+        ],
+    )
+    agg = ku.aggregate_usage_from_wire(wire)
+    assert agg["inputOther"] == 300  # noqa: PLR2004
+    assert agg["output"] == 130  # noqa: PLR2004
+    assert agg["inputCacheRead"] == 30  # noqa: PLR2004
+    assert agg["inputCacheCreation"] == 5  # noqa: PLR2004
+    assert agg["turns"] == 2  # noqa: PLR2004
+
+
+def test_aggregate_ignores_non_turn_scope_and_bad_lines(tmp_path: Path) -> None:
+    wire = tmp_path / "wire.jsonl"
+    _write_jsonl(
+        wire,
+        [
+            "not json",
+            _usage_record(input_other=5, output=1, scope="session"),
+            _usage_record(input_other=10, output=5),
+        ],
+    )
+    agg = ku.aggregate_usage_from_wire(wire)
+    assert agg["inputOther"] == 10  # noqa: PLR2004
+    assert agg["turns"] == 1
+
+
+def test_aggregate_zero_for_missing_log(tmp_path: Path) -> None:
+    agg = ku.aggregate_usage_from_wire(tmp_path / "nope.jsonl")
+    assert agg["turns"] == 0
+    assert all(v == 0 for k, v in agg.items() if k != "turns")
+
+
+# ---------------------------------------------------------------------------
+# capture_run_usage / main
+# ---------------------------------------------------------------------------
+
+
+def test_capture_run_usage_writes_usage_json(tmp_path: Path) -> None:
+    home = tmp_path / ".kimi-code"
+    session_dir = home / "sessions" / "wd_myrepo_hash1" / "session_abc"
+    (session_dir / "agents" / "main").mkdir(parents=True)
+    wire = session_dir / "agents" / "main" / "wire.jsonl"
+    _write_jsonl(wire, [_usage_record(input_other=100, output=50, cache_read=10)])
+
+    run_log = tmp_path / "run.jsonl"
+    _write_jsonl(run_log, [_resume_hint("session_abc")])
+
+    out = tmp_path / "usage.json"
+    tokens = ku.capture_run_usage(
+        run_log=run_log,
+        workdir="/data/workspaces/myrepo",
+        model="kimi-code/k3",
+        out_path=out,
+        kimi_code_home=home,
+    )
+    assert tokens == (100, 50, 10, 0)
+    data = json.loads(out.read_text())
+    assert data["model"] == "kimi-code/k3"
+    assert data["tokens_input"] == 100  # noqa: PLR2004
+    assert data["tokens_output"] == 50  # noqa: PLR2004
+    assert data["tokens_cache_read"] == 10  # noqa: PLR2004
+    assert data["turns"] == 1
+    assert data["cost_usd"] > 0.0
+
+
+def test_capture_run_usage_zero_when_no_session_found(tmp_path: Path) -> None:
+    home = tmp_path / ".kimi-code"
+    run_log = tmp_path / "run.jsonl"
+    run_log.write_text("", encoding="utf-8")
+    out = tmp_path / "usage.json"
+    tokens = ku.capture_run_usage(
+        run_log=run_log,
+        workdir="/data/workspaces/myrepo",
+        model="kimi-code/k3",
+        out_path=out,
+        kimi_code_home=home,
+    )
+    assert tokens == (0, 0, 0, 0)
+    data = json.loads(out.read_text())
+    assert data["tokens_input"] == 0
+    assert data["turns"] == 0
+
+
+def test_main_writes_usage_file(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    home = tmp_path / ".kimi-code"
+    session_dir = home / "sessions" / "wd_myrepo_hash1" / "session_abc"
+    (session_dir / "agents" / "main").mkdir(parents=True)
+    wire = session_dir / "agents" / "main" / "wire.jsonl"
+    _write_jsonl(wire, [_usage_record(input_other=200, output=100)])
+
+    run_log = tmp_path / "run.jsonl"
+    _write_jsonl(run_log, [_resume_hint("session_abc")])
+
+    out = tmp_path / "usage.json"
+    monkeypatch.setattr(ku, "USAGE_OUT_PATH", out)
+    monkeypatch.setattr(ku, "KIMI_CODE_HOME", home)
+    monkeypatch.setenv("ROBOCO_KIMI_RUN_LOG", str(run_log))
+    monkeypatch.setenv("ROBOCO_KIMI_WORKDIR", "/data/workspaces/myrepo")
+    monkeypatch.setenv("ROBOCO_AGENT_MODEL", "kimi-code/k3")
+
+    assert ku.main() == 0
+    data = json.loads(out.read_text())
+    assert data["tokens_input"] == 200  # noqa: PLR2004
+    assert data["tokens_output"] == 100  # noqa: PLR2004
+
+
+def test_main_warns_when_run_log_env_missing(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    monkeypatch.delenv("ROBOCO_KIMI_RUN_LOG", raising=False)
+    with caplog.at_level("WARNING", logger="roboco.llm.providers.kimi_cli_usage"):
+        assert ku.main() == 0
+    assert any("ROBOCO_KIMI_RUN_LOG" in r.message for r in caplog.records)

tests/unit/llm/test_providers.py

@@ -21,6 +21,7 @@ from roboco.llm.providers import (
     ClaudeCodeProvider,
     CodexCliProvider,
     GrokCliProvider,
+    KimiCliProvider,
     ProviderError,
     ProviderNotRegisteredError,
     ProviderRegistry,
@@ -48,6 +49,14 @@ def _isolate_codex_auth(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path
     return codex_dir
 
 
+@pytest.fixture(autouse=True)
+def _isolate_kimi_auth(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """Point KIMI_AUTH_HOST_PATH at a fresh tmp dir (parity with codex above)."""
+    kimi_dir = tmp_path / "kimi-auth"
+    monkeypatch.setattr("roboco.llm.providers.kimi.KIMI_AUTH_HOST_PATH", str(kimi_dir))
+    return kimi_dir
+
+
 def _config(
     *,
     agent_id: str = "be-dev-1",
@@ -99,6 +108,9 @@ class _FakeHost:
     def _ensure_codex_usage_dir(self, agent_id: str) -> None:
         self.data_dirs_ensured.append(agent_id)
 
+    def _ensure_kimi_usage_dir(self, agent_id: str) -> None:
+        self.data_dirs_ensured.append(agent_id)
+
     def _resolve_host_paths(
         self, config: OrchestratorAgentConfig, agent_settings_path: Path | None
     ) -> dict[str, str | None]:
@@ -109,6 +121,7 @@ class _FakeHost:
             "settings": str(agent_settings_path) if agent_settings_path else None,
             "grok_usage": f"/host/data/grok-usage/{config.agent_id}",
             "codex_usage": f"/host/data/codex-usage/{config.agent_id}",
+            "kimi_usage": f"/host/data/kimi-usage/{config.agent_id}",
         }
 
     def _build_mount_args(
@@ -477,6 +490,157 @@ async def test_codex_spawn_raises_on_docker_failure() -> None:
         await provider.spawn(_codex_config())
 
 
+# ---------------------------------------------------------------------------
+# KimiCliProvider
+# ---------------------------------------------------------------------------
+
+
+def _kimi_config(
+    *,
+    agent_id: str = "be-dev-1",
+    provider_base_url: str | None = "https://api.x.ai/v1",
+    provider_auth_token: str | None = "should-not-leak",
+    mcp_config_path: Path | None = Path("/host/mcp-configs/be-dev-1.json"),
+) -> OrchestratorAgentConfig:
+    return OrchestratorAgentConfig(
+        agent_id=agent_id,
+        blueprint_path=Path("/app/system-prompt.md"),
+        model="kimi-code/k3",
+        mcp_config_path=mcp_config_path,
+        claude_session_id="sess-1",
+        provider_type="kimi",
+        provider_base_url=provider_base_url,
+        provider_auth_token=provider_auth_token,
+    )
+
+
+async def test_kimi_spawn_requires_mcp_config() -> None:
+    provider = KimiCliProvider(_FakeHost())
+    with pytest.raises(ProviderError, match="MCP config"):
+        await provider.spawn(_kimi_config(mcp_config_path=None))
+
+
+async def test_kimi_spawn_does_not_require_api_key() -> None:
+    # Subscription auth (mounted ~/.kimi-code) — a missing provider key is fine.
+    host = _FakeHost()
+    provider = KimiCliProvider(host)
+    with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())):
+        result = await provider.spawn(_kimi_config(provider_auth_token=None))
+    assert result.instance_id == "roboco-agent-be-dev-1"
+
+
+async def test_kimi_spawn_no_leaked_key_and_no_anthropic_leak() -> None:
+    host = _FakeHost()
+    provider = KimiCliProvider(host, image="roboco-agent-kimi:test")
+    with patch(
+        "asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
+    ) as exec_mock:
+        await provider.spawn(_kimi_config(), initial_prompt="do the work")
+    cmd = list(exec_mock.call_args.args)
+    assert not any(c.startswith("MOONSHOT_API_KEY=") for c in cmd)
+    # The provider endpoint must NOT be injected as an Anthropic var.
+    assert not any(c.startswith("ANTHROPIC_BASE_URL=") for c in cmd)
+    assert not any(c.startswith("ANTHROPIC_AUTH_TOKEN=") for c in cmd)
+    assert host.mount_config is not None
+    assert host.mount_config.provider_base_url is None
+    assert host.mount_config.provider_auth_token is None
+
+
+async def test_kimi_spawn_wires_gateway_env_and_image_last() -> None:
+    host = _FakeHost()
+    provider = KimiCliProvider(host, image="roboco-agent-kimi:test")
+    with patch(
+        "asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
+    ) as exec_mock:
+        result = await provider.spawn(_kimi_config())
+    cmd = list(exec_mock.call_args.args)
+    assert "ROBOCO_MCP_CONFIG=/app/mcp-config.json" in cmd
+    assert "ROBOCO_AGENT_ID=be-dev-1" in cmd
+    assert "ROBOCO_AGENT_MODEL=kimi-code/k3" in cmd
+    # Usage capture: per-agent data dir mounted + the entrypoint's usage file.
+    assert host.data_dirs_ensured == ["be-dev-1"]
+    assert "/host/data/kimi-usage/be-dev-1:/home/agent/.kimi-usage" in cmd
+    assert "ROBOCO_KIMI_USAGE_FILE=/home/agent/.kimi-usage/usage.json" in cmd
+    assert "ROBOCO_AGENT_TOKEN=hmac-be-dev-1" in cmd
+    assert cmd[-1] == "roboco-agent-kimi:test"
+    assert host.removed == ["roboco-agent-be-dev-1"]
+    assert host.remove_stop_reasons == ["pre_spawn_stale_clear"]
+    assert result == SpawnResult(
+        instance_id="roboco-agent-be-dev-1",
+        extra={"container_id": "cid", "model": "kimi-code/k3"},
+    )
+
+
+async def test_kimi_spawn_mounts_auth_when_present(_isolate_kimi_auth: Path) -> None:
+    creds_dir = _isolate_kimi_auth / "credentials"
+    creds_dir.mkdir(parents=True, exist_ok=True)
+    (creds_dir / "kimi-code.json").write_text("{}", encoding="utf-8")
+    host = _FakeHost()
+    provider = KimiCliProvider(host)
+    with patch(
+        "asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
+    ) as exec_mock:
+        await provider.spawn(_kimi_config())
+    cmd = list(exec_mock.call_args.args)
+    # Mount the host ~/.kimi-code DIRECTORY read-write (rotation-with-grace,
+    # not truly reusable — every container must share ONE chain with the
+    # host, not a private copy) — the entrypoint symlinks credentials/ and
+    # oauth/ forward into a container-local, writable ~/.kimi-code.
+    expected = f"{_isolate_kimi_auth}:/home/agent/.kimi-code-auth"
+    assert expected in cmd
+
+
+async def test_kimi_spawn_omits_auth_mount_when_absent() -> None:
+    host = _FakeHost()
+    provider = KimiCliProvider(host)
+    with patch(
+        "asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
+    ) as exec_mock:
+        await provider.spawn(_kimi_config())
+    cmd = list(exec_mock.call_args.args)
+    assert not any("/home/agent/.kimi-code-auth" in c for c in cmd)
+
+
+async def test_kimi_spawn_warns_when_auth_absent(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    caplog.set_level("WARNING", logger="roboco.llm.providers.kimi")
+    host = _FakeHost()
+    provider = KimiCliProvider(host)
+    with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())):
+        await provider.spawn(_kimi_config())
+    warnings = [r for r in caplog.records if r.levelname == "WARNING"]
+    assert warnings, "expected a spawn-time WARNING for the missing host credential"
+    msg = warnings[0].getMessage()
+    assert "kimi-code.json" in msg
+    assert "kimi login" in msg
+
+
+async def test_kimi_spawn_prompt_is_injection_safe() -> None:
+    host = _FakeHost()
+    provider = KimiCliProvider(host)
+    nasty = "--model evil --session-id pwned"
+    with patch(
+        "asyncio.create_subprocess_exec", AsyncMock(return_value=_proc())
+    ) as exec_mock:
+        await provider.spawn(_kimi_config(), initial_prompt=nasty)
+    cmd = list(exec_mock.call_args.args)
+    assert f"ROBOCO_INITIAL_PROMPT={nasty}" in cmd
+    assert nasty not in cmd
+
+
+async def test_kimi_spawn_raises_on_docker_failure() -> None:
+    provider = KimiCliProvider(_FakeHost())
+    with (
+        patch(
+            "asyncio.create_subprocess_exec",
+            AsyncMock(return_value=_proc(returncode=1, stderr=b"boom")),
+        ),
+        pytest.raises(ProviderError, match="boom"),
+    ):
+        await provider.spawn(_kimi_config())
+
+
 # ---------------------------------------------------------------------------
 # ClaudeCodeProvider
 # ---------------------------------------------------------------------------

tests/unit/runtime/test_interactive_provider_guard.py

@@ -1,8 +1,9 @@
-"""Codex (OPENAI) and Gemini (GEMINI) are V1 delivery-roles-only — neither has
-an interactive-session driver image (unlike GROK's dedicated
-GROK_PROMPTER_IMAGE / GROK_SECRETARY_IMAGE). Routing either to the persistent
-Intake/Secretary agent must refuse loudly instead of silently falling through
-to the plain Claude SDK-driver image with a mismatched provider env.
+"""Codex (OPENAI), Gemini (GEMINI), and Kimi (KIMI) are V1 delivery-roles-only
+— none has an interactive-session driver image (unlike GROK's dedicated
+GROK_PROMPTER_IMAGE / GROK_SECRETARY_IMAGE). Routing any of them to the
+persistent Intake/Secretary agent must refuse loudly instead of silently
+falling through to the plain Claude SDK-driver image with a mismatched
+provider env.
 """
 
 from __future__ import annotations
@@ -67,7 +68,9 @@ class TestRejectInteractiveUnsupportedProvider:
         un-exempt a chat."""
         assert set(INTERACTIVE_AGENT_SLUGS) == {INTAKE_AGENT_ID, SECRETARY_AGENT_ID}
 
-    @pytest.mark.parametrize("provider", [ModelProvider.OPENAI, ModelProvider.GEMINI])
+    @pytest.mark.parametrize(
+        "provider", [ModelProvider.OPENAI, ModelProvider.GEMINI, ModelProvider.KIMI]
+    )
     def test_raises_for_delivery_only_providers(self, provider: ModelProvider) -> None:
         with pytest.raises(RuntimeError, match="delivery-roles-only"):
             _reject_interactive_unsupported_provider(INTAKE_AGENT_ID, provider)
@@ -93,7 +96,9 @@ class TestRejectInteractiveUnsupportedProvider:
 
 
 class TestIntakeSpawnRefusesDeliveryOnlyProvider:
-    @pytest.mark.parametrize("provider", [ModelProvider.OPENAI, ModelProvider.GEMINI])
+    @pytest.mark.parametrize(
+        "provider", [ModelProvider.OPENAI, ModelProvider.GEMINI, ModelProvider.KIMI]
+    )
     @pytest.mark.asyncio
     async def test_refuses_before_any_container_work(
         self, monkeypatch: pytest.MonkeyPatch, provider: ModelProvider
@@ -149,7 +154,9 @@ class TestIntakeSpawnRefusesDeliveryOnlyProvider:
 
 
 class TestSecretarySpawnRefusesDeliveryOnlyProvider:
-    @pytest.mark.parametrize("provider", [ModelProvider.OPENAI, ModelProvider.GEMINI])
+    @pytest.mark.parametrize(
+        "provider", [ModelProvider.OPENAI, ModelProvider.GEMINI, ModelProvider.KIMI]
+    )
     @pytest.mark.asyncio
     async def test_refuses_before_any_container_work(
         self, monkeypatch: pytest.MonkeyPatch, provider: ModelProvider

tests/unit/runtime/test_kimi_rate_limit.py

@@ -0,0 +1,156 @@
+"""KIMI 429/auth parking: same exit-code convention as codex/grok, scoped to
+ModelProvider.KIMI so a numeric-code collision with another provider's crash
+can never mis-park (see ``_KIMI_RATE_LIMIT_EXIT_CODE`` / ``_KIMI_AUTH_EXIT_CODE``
+in ``roboco.runtime.orchestrator``).
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+from roboco.models.runtime import AgentInstance
+from roboco.runtime.orchestrator import (
+    _KIMI_AUTH_EXIT_CODE,
+    _KIMI_RATE_LIMIT_EXIT_CODE,
+    AgentOrchestrator,
+    AgentState,
+)
+
+
+def _kimi_instance(provider_type: str = "kimi") -> AgentInstance:
+    cfg = type("C", (), {"provider_type": provider_type, "model": "kimi-code/k3"})()
+    inst = AgentInstance(agent_id="be-dev-1", state=AgentState.ACTIVE, config=cfg)
+    inst.current_task_id = "task-1"
+    inst.container_id = "cid"
+    return inst
+
+
+class _FakeTracker:
+    def __init__(self) -> None:
+        self.activated_with: dict[str, object] | None = None
+
+    async def activate(
+        self,
+        *,
+        retry_after: float,
+        affected_agents: list[str],
+        kind: str = "rate_limited",
+    ) -> None:
+        self.activated_with = {
+            "retry_after": retry_after,
+            "affected_agents": affected_agents,
+            "kind": kind,
+        }
+
+
+def test_is_kimi_rate_limit_exit() -> None:
+    inst = _kimi_instance()
+    assert AgentOrchestrator._is_kimi_rate_limit_exit(inst, _KIMI_RATE_LIMIT_EXIT_CODE)
+    assert not AgentOrchestrator._is_kimi_rate_limit_exit(inst, 0)
+    assert not AgentOrchestrator._is_kimi_rate_limit_exit(inst, 1)
+    # A codex exit at the SAME numeric code must NOT be classified as kimi.
+    assert not AgentOrchestrator._is_kimi_rate_limit_exit(
+        _kimi_instance(provider_type="openai"), _KIMI_RATE_LIMIT_EXIT_CODE
+    )
+    assert not AgentOrchestrator._is_kimi_rate_limit_exit(
+        _kimi_instance(provider_type="anthropic"), _KIMI_RATE_LIMIT_EXIT_CODE
+    )
+
+
+def test_is_kimi_auth_exit() -> None:
+    inst = _kimi_instance()
+    assert AgentOrchestrator._is_kimi_auth_exit(inst, _KIMI_AUTH_EXIT_CODE)
+    assert not AgentOrchestrator._is_kimi_auth_exit(inst, 0)
+    assert not AgentOrchestrator._is_kimi_auth_exit(inst, 1)
+    assert not AgentOrchestrator._is_kimi_auth_exit(
+        _kimi_instance(provider_type="openai"), _KIMI_AUTH_EXIT_CODE
+    )
+
+
+@pytest.mark.asyncio
+async def test_park_kimi_rate_limited_activates_and_offlines(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    orch._waiting_records = {}
+    orch._rate_limit_ceo_notified = set()
+    inst = _kimi_instance()
+    inst.error_count = 2  # pretend prior crashes — parking must NOT count one
+    tracker = _FakeTracker()
+    monkeypatch.setattr(orch, "_make_tracker", lambda _p: tracker)
+    finalize = AsyncMock()
+    monkeypatch.setattr(orch, "_finalize_spawn_session", finalize)
+    monkeypatch.setattr(orch, "_persist_waiting_record", AsyncMock())
+
+    await orch._park_kimi_rate_limited("be-dev-1", inst)
+
+    finalize.assert_awaited_once()
+    assert inst.state == AgentState.OFFLINE
+    assert inst.container_id is None
+    assert inst.error_count == 0  # a 429 is not a crash
+    assert tracker.activated_with == {
+        "retry_after": pytest.approx(60.0),
+        "affected_agents": ["be-dev-1"],
+        "kind": "rate_limited",
+    }
+
+
+@pytest.mark.asyncio
+async def test_park_kimi_auth_unavailable_activates_with_auth_missing_kind(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    orch._waiting_records = {}
+    orch._rate_limit_ceo_notified = set()
+    inst = _kimi_instance()
+    inst.error_count = 2
+    tracker = _FakeTracker()
+    monkeypatch.setattr(orch, "_make_tracker", lambda _p: tracker)
+    monkeypatch.setattr(orch, "_finalize_spawn_session", AsyncMock())
+    monkeypatch.setattr(orch, "_persist_waiting_record", AsyncMock())
+
+    await orch._park_kimi_auth_unavailable("be-dev-1", inst)
+
+    assert inst.state == AgentState.OFFLINE
+    assert inst.container_id is None
+    assert inst.error_count == 0
+    assert tracker.activated_with == {
+        "retry_after": pytest.approx(60.0),
+        "affected_agents": ["be-dev-1"],
+        "kind": "auth_missing",
+    }
+
+
+@pytest.mark.asyncio
+async def test_handle_stopped_container_parks_on_kimi_429(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    inst = _kimi_instance()
+    park = AsyncMock()
+    finalize = AsyncMock()
+    monkeypatch.setattr(orch, "_park_kimi_rate_limited", park)
+    monkeypatch.setattr(orch, "_finalize_spawn_session", finalize)
+
+    await orch._handle_stopped_container("be-dev-1", inst, _KIMI_RATE_LIMIT_EXIT_CODE)
+
+    park.assert_awaited_once_with("be-dev-1", inst)
+    finalize.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_handle_stopped_container_parks_on_kimi_auth_exit(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    inst = _kimi_instance()
+    park = AsyncMock()
+    finalize = AsyncMock()
+    monkeypatch.setattr(orch, "_park_kimi_auth_unavailable", park)
+    monkeypatch.setattr(orch, "_finalize_spawn_session", finalize)
+
+    await orch._handle_stopped_container("be-dev-1", inst, _KIMI_AUTH_EXIT_CODE)
+
+    park.assert_awaited_once_with("be-dev-1", inst)
+    finalize.assert_not_awaited()

tests/unit/runtime/test_kimi_usage_finalize.py

@@ -0,0 +1,144 @@
+"""KIMI agents capture real input/output/cache-split token usage from their
+captured ``usage.json`` — Kimi's wire.jsonl carries a genuine, already-disjoint
+4-bucket split (see ``kimi_cli_usage``), so finalize must return the real
+4-tuple instead of folding everything into output.
+"""
+
+from __future__ import annotations
+
+import json
+import tempfile
+from typing import TYPE_CHECKING
+
+import httpx
+import pytest
+from roboco.models.runtime import AgentInstance
+from roboco.runtime import orchestrator as orch_mod
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _write_usage(path: Path, **fields: object) -> None:
+    payload = {
+        "model": "kimi-code/k3",
+        "tokens_input": 0,
+        "tokens_output": 0,
+        "tokens_cache_read": 0,
+        "tokens_cache_write": 0,
+        "cost_usd": 0.0,
+        "turns": 1,
+        **fields,
+    }
+    path.write_text(json.dumps(payload), encoding="utf-8")
+
+
+def test_kimi_usage_returns_real_split(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    usage = tmp_path / "usage.json"
+    _write_usage(
+        usage, tokens_input=300, tokens_output=130, tokens_cache_read=30, turns=2
+    )
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    monkeypatch.setattr(
+        orch, "_kimi_usage_json", lambda _aid: json.loads(usage.read_text())
+    )
+
+    expected_turns = 2
+    assert orch._kimi_usage_tokens("be-dev-1") == (300, 130, 30, 0)
+    assert orch._kimi_usage_turns("be-dev-1") == expected_turns
+
+
+def test_kimi_usage_zero_when_store_missing(monkeypatch: pytest.MonkeyPatch) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    monkeypatch.setattr(orch, "_kimi_usage_json", lambda _aid: None)
+    assert orch._kimi_usage_tokens("be-dev-1") == (0, 0, 0, 0)
+    assert orch._kimi_usage_turns("be-dev-1") == 0
+
+
+@pytest.mark.asyncio
+async def test_resolve_final_usage_routes_kimi_to_usage_json(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    monkeypatch.setattr(
+        orch,
+        "_kimi_usage_json",
+        lambda _aid: {
+            "tokens_input": 12,
+            "tokens_output": 34,
+            "tokens_cache_read": 5,
+            "tokens_cache_write": 1,
+        },
+    )
+    cfg = type("C", (), {"provider_type": "kimi"})()
+    orch._instances = {"be-dev-1": AgentInstance(agent_id="be-dev-1", config=cfg)}
+
+    assert await orch._resolve_final_token_usage("be-dev-1") == (12, 34, 5, 1)
+
+
+@pytest.mark.asyncio
+async def test_resolve_final_turns_tools_routes_kimi_to_usage_json(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    monkeypatch.setattr(orch, "_kimi_usage_turns", lambda _aid: 3)
+    cfg = type("C", (), {"provider_type": "kimi"})()
+    orch._instances = {"be-dev-1": AgentInstance(agent_id="be-dev-1", config=cfg)}
+
+    # Kimi has no tool-call signal — tool_calls stays 0.
+    assert await orch._resolve_final_turns_tools("be-dev-1") == (3, 0)
+
+
+@pytest.mark.asyncio
+async def test_resolve_active_tokens_routes_kimi_to_usage_json(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    monkeypatch.setattr(
+        orch,
+        "_kimi_usage_json",
+        lambda _aid: {"tokens_input": 12, "tokens_output": 34},
+    )
+    cfg = type("C", (), {"provider_type": "kimi"})()
+    orch._instances = {"be-dev-1": AgentInstance(agent_id="be-dev-1", config=cfg)}
+    async with httpx.AsyncClient() as client:
+        assert await orch._resolve_active_tokens(client, "be-dev-1") == (12, 34, 0, 0)
+
+
+def test_kimi_usage_dir_branches_compose_vs_local(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(orch_mod, "PROJECT_HOST_PATH", "")
+    local = AgentOrchestrator._kimi_usage_dir("be-dev-1")
+    assert "roboco-kimi-usage" in str(local)
+    assert local.name == "be-dev-1"
+
+    monkeypatch.setattr(orch_mod, "PROJECT_HOST_PATH", "/volume1/roboco")
+    monkeypatch.setattr(orch_mod, "KIMI_USAGE_DATA_DIR", "/data/kimi-usage")
+    assert str(AgentOrchestrator._kimi_usage_dir("be-dev-1")) == (
+        "/data/kimi-usage/be-dev-1"
+    )
+
+
+@pytest.mark.parametrize(
+    "bad",
+    ["..", ".", "../etc", "a/b", "a\\b", "", "be-dev-1/../x", "x\x00y"],
+)
+def test_kimi_usage_dir_rejects_path_traversal(bad: str) -> None:
+    with pytest.raises(ValueError, match="unsafe agent id"):
+        AgentOrchestrator._kimi_usage_dir(bad)
+
+
+def test_kimi_usage_json_reads_the_real_local_dir(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(orch_mod, "PROJECT_HOST_PATH", "")
+    monkeypatch.setattr(tempfile, "gettempdir", lambda: str(tmp_path))
+    udir = tmp_path / "roboco-kimi-usage" / "be-dev-1"
+    udir.mkdir(parents=True)
+    _write_usage(udir / "usage.json", tokens_input=55, tokens_output=10)
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    assert orch._kimi_usage_tokens("be-dev-1") == (55, 10, 0, 0)