"""Unit tests for the kimi-native (terminal-injection) harness. Covers the executor's text extraction + capability flags, the tmux bridge's pure helpers (paste-payload encoding, bridge dir, spawn env, tmux.json round-trip), and harness registration. The live tmux injection is exercised by the e2e gate, not here, so these need no tmux or kimi binary. Unlike cursor-native, kimi-native has NO MCP plumbing (upstream kimi has no per-spawn MCP config), so the MCP-config tests have no analogue here. """ from __future__ import annotations from pathlib import Path import pytest from omnigent import kimi_native_bridge from omnigent.inner.kimi_native_executor import ( KimiNativeExecutor, _content_to_text, _latest_user_text, ) from omnigent.kimi_native_bridge import ( APPROVE_KEY, BRIDGE_DIR_ENV_VAR, DENY_KEY, _paste_payload_bytes, bridge_dir_for_session_id, build_kimi_native_spawn_env, inject_approval_keystroke, read_tmux_info, write_tmux_target, ) class TestContentExtraction: def test_string_content(self, tmp_path: Path) -> None: assert _content_to_text("hello", tmp_path) == "hello" def test_input_text_blocks(self, tmp_path: Path) -> None: content = [ {"type": "input_text", "text": "one"}, {"type": "text", "text": "two"}, # invalid data URI -> materialize_attachment returns None -> no line {"type": "input_image", "image_url": "data:..."}, ] assert _content_to_text(content, tmp_path) == "one\n\ntwo" def test_real_image_attachment_materialized(self, tmp_path: Path) -> None: # a tiny valid base64 PNG data URI should be written to disk + referenced png = ( "data:image/png;base64," "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==" ) out = _content_to_text([{"type": "input_image", "image_url": png}], tmp_path) assert out.startswith("[Attached: ") assert str(tmp_path) in out def test_empty_and_none(self, tmp_path: Path) -> None: assert _content_to_text(None, tmp_path) == "" assert _content_to_text([], tmp_path) == "" def test_latest_user_text(self, tmp_path: Path) -> None: messages = [ {"role": "user", "content": "first"}, {"role": "assistant", "content": "ok"}, {"role": "user", "content": "second"}, ] assert _latest_user_text(messages, tmp_path) == "second" assert _latest_user_text([{"role": "assistant", "content": "x"}], tmp_path) == "" class TestExecutorCapabilities: def test_capability_flags(self, tmp_path: Path) -> None: ex = KimiNativeExecutor(bridge_dir=tmp_path) # Output is shown by the embedded terminal, not streamed by the executor. assert ex.supports_streaming() is False # Web-UI messages can be injected mid-turn (steering). assert ex.supports_live_message_queue() is True class TestPastePayload: def test_newlines_become_cr(self) -> None: assert _paste_payload_bytes("a\nb") == b"a\rb" assert _paste_payload_bytes("a\r\nb") == b"a\rb" assert _paste_payload_bytes("a\rb") == b"a\rb" def test_tab_kept_other_control_dropped(self) -> None: # tab kept (0x09), ESC (0x1b) and BEL (0x07) dropped. assert _paste_payload_bytes("a\tb\x1b\x07c") == b"a\tbc" def test_unicode_passthrough(self) -> None: assert _paste_payload_bytes("café") == "café".encode() class TestBridge: def test_bridge_dir_is_deterministic_and_session_scoped(self) -> None: a1 = bridge_dir_for_session_id("conv_a") a2 = bridge_dir_for_session_id("conv_a") b = bridge_dir_for_session_id("conv_b") assert a1 == a2 assert a1 != b assert "kimi-native" in str(a1) def test_spawn_env_carries_bridge_dir(self) -> None: env = build_kimi_native_spawn_env("conv_xyz") assert env[BRIDGE_DIR_ENV_VAR] == str(bridge_dir_for_session_id("conv_xyz")) # Only the bridge dir is emitted (no MCP / active-session guard env). assert list(env) == [BRIDGE_DIR_ENV_VAR] def test_tmux_target_round_trip(self, tmp_path: Path) -> None: write_tmux_target(tmp_path, socket_path=Path("/tmp/x/tmux.sock"), tmux_target="main") info = read_tmux_info(tmp_path) assert info == {"socket_path": "/tmp/x/tmux.sock", "tmux_target": "main"} def test_read_tmux_info_missing(self, tmp_path: Path) -> None: assert read_tmux_info(tmp_path) is None class TestApprovalKeystroke: """`inject_approval_keystroke` types the option digit + Enter, guarded by the permission-menu marker so a stray verdict can't leak a keystroke.""" def _stub_tmux( self, monkeypatch: pytest.MonkeyPatch, *, pane: str, alive: bool = True ) -> list[tuple[str, ...]]: sent: list[tuple[str, ...]] = [] monkeypatch.setattr( kimi_native_bridge, "_wait_for_tmux_info", lambda bridge_dir, *, timeout_s: {"socket_path": "/s", "tmux_target": "main"}, ) monkeypatch.setattr(kimi_native_bridge, "_session_alive", lambda s, t: alive) monkeypatch.setattr(kimi_native_bridge, "_capture_pane", lambda s, t: pane) monkeypatch.setattr( kimi_native_bridge, "_run_tmux", lambda socket_path, *args: sent.append(args), ) return sent def test_injects_digit_and_enter_when_menu_present( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: sent = self._stub_tmux(monkeypatch, pane="▶ 1. Approve once\n 3. Reject") assert inject_approval_keystroke(tmp_path, key=APPROVE_KEY) is True assert sent == [ ("send-keys", "-t", "main", APPROVE_KEY), ("send-keys", "-t", "main", "Enter"), ] def test_deny_key_selects_reject( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: sent = self._stub_tmux(monkeypatch, pane="▶ 1. Approve once\n 3. Reject") assert inject_approval_keystroke(tmp_path, key=DENY_KEY) is True assert sent[0] == ("send-keys", "-t", "main", DENY_KEY) def test_skips_when_menu_absent(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: # Prompt already answered in the terminal → marker gone → no keystroke. sent = self._stub_tmux(monkeypatch, pane="● Hello! How can I help?") assert inject_approval_keystroke(tmp_path, key=APPROVE_KEY) is False assert sent == [] def test_skips_when_tui_exited(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: sent = self._stub_tmux(monkeypatch, pane="▶ 1. Approve once", alive=False) assert inject_approval_keystroke(tmp_path, key=APPROVE_KEY) is False assert sent == [] class TestSettlePaneReadiness: """``_settle_pane`` must recognize the real kimi TUI chrome so it returns on the first capture — a wrong marker silently burns the full readiness timeout on every web→TUI injection (the original web→TUI latency bug).""" def test_marker_matches_live_kimi_footer(self) -> None: # Footer chrome captured verbatim from a live K2.7 session. footer = ( " K2.7 Code thinking ~/omnigent pr521-kimi-native [+61 -8]" ' ask Kimi to schedule tasks, e.g. "remind me at 5pm"\n' " context: 6.5% (17.0k/262.1k)" ) assert any(m in footer for m in kimi_native_bridge._INPUT_READY_MARKERS) # The cursor-native strings carried over unverified never appeared. assert "Plan, search, build" not in footer assert "Add a follow-up" not in footer def test_settle_returns_on_first_capture_when_ready( self, monkeypatch: pytest.MonkeyPatch ) -> None: captures = {"n": 0} def _capture(_s: str, _t: str) -> str: captures["n"] += 1 return "context: 6.5% (17.0k/262.1k)" monkeypatch.setattr(kimi_native_bridge, "_capture_pane", _capture) # If the marker fails to match, this would loop until the deadline; a # tiny timeout keeps the test fast either way, but it must return after # exactly one capture with no sleep. slept: list[float] = [] monkeypatch.setattr(kimi_native_bridge.time, "sleep", lambda s: slept.append(s)) kimi_native_bridge._settle_pane("/s", "main", timeout_s=30.0) assert captures["n"] == 1 assert slept == [] class TestRegistration: def test_harness_is_registered(self) -> None: from omnigent.runtime.harnesses import _HARNESS_MODULES assert _HARNESS_MODULES["kimi-native"] == "omnigent.inner.kimi_native_harness" def test_harness_is_allowlisted(self) -> None: from omnigent.spec._omnigent_compat import OMNIGENT_HARNESSES assert "kimi-native" in OMNIGENT_HARNESSES def test_kimi_native_is_terminal_native(self) -> None: # kimi-native launches the kimi TUI in an omnigent terminal (like # claude/codex/cursor-native), so the runner must treat it as a native # terminal harness. from omnigent.harness_aliases import is_native_harness assert is_native_harness("kimi-native") is True assert is_native_harness("native-kimi") is True def test_native_coding_agent_record(self) -> None: from omnigent.native_coding_agents import native_coding_agent_for_harness agent = native_coding_agent_for_harness("kimi-native") assert agent is not None assert agent.terminal_name == "kimi" assert agent.display_name == "Kimi" def test_distinct_from_headless_kimi_harness(self) -> None: # The bare ``kimi`` harness is the headless SDK path; ``kimi-native`` is # the TUI path. They must resolve to different harness modules. from omnigent.runtime.harnesses import _HARNESS_MODULES assert _HARNESS_MODULES["kimi"] != _HARNESS_MODULES["kimi-native"]