fix(kimi): cover remaining fixed-temperature bypasses

This commit is contained in:
helix4u
2026-04-17 20:39:24 -06:00
committed by Teknium
parent 53e4a2f2c6
commit 148459716c
7 changed files with 145 additions and 20 deletions

View File

@@ -0,0 +1,28 @@
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
def test_run_task_forces_kimi_fixed_temperature():
with patch("openai.OpenAI") as mock_openai:
client = MagicMock()
client.chat.completions.create.return_value = SimpleNamespace(
choices=[SimpleNamespace(message=SimpleNamespace(content="done", tool_calls=[]))]
)
mock_openai.return_value = client
from mini_swe_runner import MiniSWERunner
runner = MiniSWERunner(
model="kimi-for-coding",
base_url="https://api.kimi.com/coding/v1",
api_key="test-key",
env_type="local",
max_iterations=1,
)
runner._create_env = MagicMock()
runner._cleanup_env = MagicMock()
result = runner.run_task("2+2")
assert result["completed"] is True
assert client.chat.completions.create.call_args.kwargs["temperature"] == 0.6

View File

@@ -31,6 +31,29 @@ def test_import_loads_env_from_hermes_home(tmp_path, monkeypatch):
assert os.getenv("OPENROUTER_API_KEY") == "from-hermes-home"
def test_generate_summary_custom_client_forces_kimi_temperature():
config = CompressionConfig(
summarization_model="kimi-for-coding",
temperature=0.3,
summary_target_tokens=100,
max_retries=1,
)
compressor = TrajectoryCompressor.__new__(TrajectoryCompressor)
compressor.config = config
compressor.logger = MagicMock()
compressor._use_call_llm = False
compressor.client = MagicMock()
compressor.client.chat.completions.create.return_value = SimpleNamespace(
choices=[SimpleNamespace(message=SimpleNamespace(content="[CONTEXT SUMMARY]: summary"))]
)
metrics = TrajectoryMetrics()
result = compressor._generate_summary("tool output", metrics)
assert result.startswith("[CONTEXT SUMMARY]:")
assert compressor.client.chat.completions.create.call_args.kwargs["temperature"] == 0.6
# ---------------------------------------------------------------------------
# CompressionConfig
# ---------------------------------------------------------------------------

View File

@@ -11,6 +11,7 @@ each asyncio.run() gets a client bound to the current loop.
"""
import types
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
@@ -113,3 +114,30 @@ class TestSourceLineVerification:
"""_get_async_client method should exist."""
src = self._read_file()
assert "def _get_async_client(self)" in src
@pytest.mark.asyncio
async def test_generate_summary_async_custom_client_forces_kimi_temperature():
from trajectory_compressor import CompressionConfig, TrajectoryCompressor, TrajectoryMetrics
config = CompressionConfig(
summarization_model="kimi-for-coding",
temperature=0.3,
summary_target_tokens=100,
max_retries=1,
)
compressor = TrajectoryCompressor.__new__(TrajectoryCompressor)
compressor.config = config
compressor.logger = MagicMock()
compressor._use_call_llm = False
async_client = MagicMock()
async_client.chat.completions.create = MagicMock(return_value=SimpleNamespace(
choices=[SimpleNamespace(message=SimpleNamespace(content="[CONTEXT SUMMARY]: summary"))]
))
compressor._get_async_client = MagicMock(return_value=async_client)
metrics = TrajectoryMetrics()
result = await compressor._generate_summary_async("tool output", metrics)
assert result.startswith("[CONTEXT SUMMARY]:")
assert async_client.chat.completions.create.call_args.kwargs["temperature"] == 0.6

View File

@@ -2,11 +2,13 @@
import ast
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch as mock_patch
import tools.approval as approval_module
from tools.approval import (
_get_approval_mode,
_smart_approve,
approve_session,
detect_dangerous_command,
is_approved,
@@ -26,6 +28,21 @@ class TestApprovalModeParsing:
assert _get_approval_mode() == "off"
class TestSmartApproval:
def test_smart_approval_uses_call_llm(self):
response = SimpleNamespace(
choices=[SimpleNamespace(message=SimpleNamespace(content="APPROVE"))]
)
with mock_patch("agent.auxiliary_client.call_llm", return_value=response) as mock_call:
result = _smart_approve("python -c \"print('hello')\"", "script execution via -c flag")
assert result == "approve"
mock_call.assert_called_once()
assert mock_call.call_args.kwargs["task"] == "approval"
assert mock_call.call_args.kwargs["temperature"] == 0
assert mock_call.call_args.kwargs["max_tokens"] == 16
class TestDetectDangerousRm:
def test_rm_rf_detected(self):
is_dangerous, key, desc = detect_dangerous_command("rm -rf /home/user")
@@ -820,4 +837,3 @@ class TestChmodExecuteCombo:
dangerous, _, _ = detect_dangerous_command(cmd)
assert dangerous is False