Nick
a65e5d44de
Module-level context_length used OllamaChatTransport unconditionally, so every cloud-pinned task logged context_discovery tier=1 HTTPError 400 and was misbudgeted at the 8192 fallback instead of the provider's real window (131072 for Verboo Code/glm-5.3-flash). is_cloud_pinned is now THE provider-prefix rule — one authority in openai_compat_transport, consumed by context discovery (new) and transport selection (deduped from rune_operator._transport_for). Red-first regression tests added. Scaffolding-derived impact: consumer rune_operator; gates suite/lint/ henxels + inference/orchestration canaries.
632 lines
24 KiB
Python
632 lines
24 KiB
Python
"""Contract tests for the bounded local-model tool runtime."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from typing import Sequence, cast
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
|
|
from rune.local_tool_runtime import ( # noqa: E402
|
|
ChatMessage,
|
|
LocalToolRuntime,
|
|
OllamaChatTransport,
|
|
)
|
|
from rune.tool_executor import TOOLS # noqa: E402
|
|
from rune.tool_protocol import OperatorResponse # noqa: E402
|
|
|
|
|
|
def native_call(call_id: str, name: str, arguments: object) -> dict:
|
|
return {
|
|
"id": call_id,
|
|
"function": {"name": name, "arguments": arguments},
|
|
}
|
|
|
|
|
|
class FakeTransport:
|
|
def __init__(
|
|
self,
|
|
responses: Sequence[object],
|
|
*,
|
|
native_tools: bool = True,
|
|
thinking: bool = True,
|
|
):
|
|
self.responses = list(responses)
|
|
self.native_tools = native_tools
|
|
self.thinking = thinking
|
|
self.calls: list[dict] = []
|
|
|
|
def supports_tools(self, model: str) -> bool:
|
|
return self.native_tools
|
|
|
|
def supports_thinking(self, model: str) -> bool:
|
|
return self.thinking
|
|
|
|
def chat(
|
|
self,
|
|
*,
|
|
model: str,
|
|
messages: list[ChatMessage],
|
|
tools: list[dict] | None,
|
|
options: dict | None = None,
|
|
think: bool | str | None = None,
|
|
) -> ChatMessage:
|
|
request = {
|
|
"model": model,
|
|
"messages": messages,
|
|
"tools": tools,
|
|
"options": options,
|
|
"think": think,
|
|
}
|
|
self.calls.append(copy.deepcopy(request))
|
|
response = self.responses.pop(0)
|
|
if isinstance(response, BaseException):
|
|
raise response
|
|
if not isinstance(response, dict):
|
|
raise TypeError("fake response must be a message object")
|
|
return cast(ChatMessage, copy.deepcopy(response))
|
|
|
|
|
|
class RecordingExecutor:
|
|
def __init__(self, results: list[dict] | None = None):
|
|
self.results = list(results or [])
|
|
self.calls: list[tuple[str, dict]] = []
|
|
|
|
def __call__(self, name: str, arguments: dict) -> dict:
|
|
self.calls.append((name, copy.deepcopy(arguments)))
|
|
if self.results:
|
|
return self.results.pop(0)
|
|
return {"success": True, "result": name}
|
|
|
|
|
|
class LocalToolRuntimeTests(unittest.TestCase):
|
|
def make_runtime(
|
|
self,
|
|
responses: Sequence[object],
|
|
*,
|
|
executor: RecordingExecutor | None = None,
|
|
native_tools: bool = True,
|
|
thinking: bool = True,
|
|
max_iterations: int = 8,
|
|
tools: list[dict] | None = None,
|
|
) -> tuple[LocalToolRuntime, FakeTransport, RecordingExecutor]:
|
|
transport = FakeTransport(responses, native_tools=native_tools, thinking=thinking)
|
|
executor = executor or RecordingExecutor()
|
|
runtime = LocalToolRuntime(
|
|
transport=transport,
|
|
executor=executor,
|
|
model="qwen3:8b",
|
|
tools=tools,
|
|
max_iterations=max_iterations,
|
|
)
|
|
return runtime, transport, executor
|
|
|
|
def test_one_native_call_appends_exact_assistant_then_tool_and_returns_final(self):
|
|
assistant = {
|
|
"role": "assistant",
|
|
"content": "",
|
|
"thinking": "I should inspect.",
|
|
"tool_calls": [native_call("1", "list_directory", {"path": "/tmp"})],
|
|
}
|
|
runtime, transport, executor = self.make_runtime([assistant, {"role": "assistant", "content": "Final answer"}])
|
|
initial: list[ChatMessage] = [{"role": "user", "content": "What is there?"}]
|
|
|
|
response = runtime.run(initial, options={"temperature": 0}, think=True)
|
|
|
|
self.assertIsInstance(response, OperatorResponse)
|
|
self.assertTrue(response.success)
|
|
self.assertEqual(response.user_text, "Final answer")
|
|
self.assertEqual(executor.calls, [("list_directory", {"path": "/tmp"})])
|
|
self.assertEqual(transport.calls[1]["messages"][1], assistant)
|
|
self.assertEqual(transport.calls[1]["messages"][2]["role"], "tool")
|
|
self.assertEqual(transport.calls[1]["messages"][2]["tool_name"], "list_directory")
|
|
self.assertEqual(transport.calls[1]["messages"][2]["tool_call_id"], "1")
|
|
self.assertEqual(
|
|
json.loads(transport.calls[1]["messages"][2]["content"]),
|
|
{"success": True, "result": "list_directory"},
|
|
)
|
|
self.assertEqual(transport.calls[0]["model"], "qwen3:8b")
|
|
self.assertEqual(transport.calls[0]["options"], {"temperature": 0})
|
|
self.assertIs(transport.calls[0]["think"], True)
|
|
self.assertTrue(transport.calls[0]["tools"])
|
|
self.assertEqual(initial, [{"role": "user", "content": "What is there?"}])
|
|
|
|
def test_non_thinking_model_does_not_receive_requested_think_parameter(self):
|
|
runtime, transport, _ = self.make_runtime(
|
|
[{"role": "assistant", "content": "plain answer"}],
|
|
thinking=False,
|
|
)
|
|
|
|
response = runtime.run([{"role": "user", "content": "hello"}], think=True)
|
|
|
|
self.assertTrue(response.success)
|
|
self.assertIsNone(transport.calls[0]["think"])
|
|
|
|
def test_ollama_transport_uses_advertised_thinking_capability_for_payload(self):
|
|
transport = OllamaChatTransport()
|
|
posted: list[tuple[str, dict]] = []
|
|
|
|
def fake_post(path: str, payload: dict) -> dict:
|
|
posted.append((path, copy.deepcopy(payload)))
|
|
if path == "/api/show":
|
|
return {"capabilities": ["completion", "tools"]}
|
|
return {"message": {"role": "assistant", "content": "ok"}}
|
|
|
|
transport._post = fake_post # type: ignore[method-assign]
|
|
|
|
self.assertFalse(transport.supports_thinking("qwen2.5:7b"))
|
|
transport.chat(
|
|
model="qwen2.5:7b",
|
|
messages=[{"role": "user", "content": "hello"}],
|
|
tools=None,
|
|
think=None,
|
|
)
|
|
|
|
chat_payload = posted[-1][1]
|
|
self.assertNotIn("think", chat_payload)
|
|
|
|
def test_two_sequential_calls_are_executed_across_model_turns(self):
|
|
runtime, transport, executor = self.make_runtime(
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [native_call("1", "list_directory", {"path": "/tmp"})],
|
|
},
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [native_call("2", "read_file", {"path": "/tmp/a"})],
|
|
},
|
|
{"role": "assistant", "content": "done"},
|
|
]
|
|
)
|
|
|
|
response = runtime.run([{"role": "user", "content": "inspect"}])
|
|
|
|
self.assertTrue(response.success)
|
|
self.assertEqual([name for name, _ in executor.calls], ["list_directory", "read_file"])
|
|
self.assertEqual(len(transport.calls), 3)
|
|
self.assertEqual([event.kind for event in response.tool_events], ["completed", "completed"])
|
|
|
|
def test_simultaneous_calls_append_one_tool_result_per_request_in_order(self):
|
|
runtime, transport, executor = self.make_runtime(
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [
|
|
native_call("1", "list_directory", {"path": "/tmp"}),
|
|
native_call("2", "read_file", {"path": "/tmp/a"}),
|
|
],
|
|
},
|
|
{"role": "assistant", "content": "done"},
|
|
]
|
|
)
|
|
|
|
response = runtime.run([{"role": "user", "content": "inspect"}])
|
|
|
|
self.assertTrue(response.success)
|
|
self.assertEqual([name for name, _ in executor.calls], ["list_directory", "read_file"])
|
|
followup = transport.calls[1]["messages"]
|
|
self.assertEqual(
|
|
[message["role"] for message in followup[-3:]],
|
|
["assistant", "tool", "tool"],
|
|
)
|
|
|
|
def test_duplicate_name_and_canonical_arguments_execute_once_per_model_turn(self):
|
|
runtime, transport, executor = self.make_runtime(
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [
|
|
native_call("1", "read_file", {"path": "/tmp/a", "limit": 5}),
|
|
native_call("2", "read_file", {"limit": 5, "path": "/tmp/a"}),
|
|
],
|
|
},
|
|
{"role": "assistant", "content": "done"},
|
|
]
|
|
)
|
|
|
|
response = runtime.run([{"role": "user", "content": "inspect"}])
|
|
|
|
self.assertTrue(response.success)
|
|
self.assertEqual(len(executor.calls), 1)
|
|
self.assertEqual(len(response.tool_events), 1)
|
|
self.assertEqual(
|
|
[m["role"] for m in transport.calls[1]["messages"][-2:]],
|
|
["assistant", "tool"],
|
|
)
|
|
|
|
def test_malformed_native_call_fails_without_executing(self):
|
|
runtime, _, executor = self.make_runtime(
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [native_call("1", "read_file", "{bad")],
|
|
}
|
|
]
|
|
)
|
|
|
|
response = runtime.run([{"role": "user", "content": "inspect"}])
|
|
|
|
self.assertFalse(response.success)
|
|
self.assertIn("malformed", response.error)
|
|
self.assertEqual(executor.calls, [])
|
|
|
|
def test_tool_error_is_returned_to_model_and_loop_continues(self):
|
|
executor = RecordingExecutor([{"success": False, "error": "permission denied"}])
|
|
runtime, transport, _ = self.make_runtime(
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [native_call("1", "read_file", {"path": "/tmp/a"})],
|
|
},
|
|
{"role": "assistant", "content": "Could not read it."},
|
|
],
|
|
executor=executor,
|
|
)
|
|
|
|
response = runtime.run([{"role": "user", "content": "inspect"}])
|
|
|
|
self.assertTrue(response.success)
|
|
self.assertEqual(response.user_text, "Could not read it.")
|
|
self.assertEqual(
|
|
json.loads(transport.calls[1]["messages"][-1]["content"])["error"],
|
|
"permission denied",
|
|
)
|
|
self.assertEqual(response.tool_events[0].kind, "error")
|
|
|
|
def test_transport_error_returns_structured_failure(self):
|
|
runtime, _, executor = self.make_runtime([RuntimeError("offline")])
|
|
|
|
response = runtime.run([{"role": "user", "content": "hello"}])
|
|
|
|
self.assertIsInstance(response, OperatorResponse)
|
|
self.assertFalse(response.success)
|
|
self.assertIn("offline", response.error)
|
|
self.assertEqual(executor.calls, [])
|
|
|
|
def test_pending_approval_returns_serializable_state_without_mutating_input(self):
|
|
executor = RecordingExecutor([{"success": True, "needs_confirm": True, "diff": "--- old\n+++ new"}])
|
|
assistant = {
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [native_call("w1", "write_file", {"path": "/tmp/a", "content": "new"})],
|
|
}
|
|
runtime, _, _ = self.make_runtime([assistant], executor=executor)
|
|
initial: list[ChatMessage] = [{"role": "user", "content": "change it"}]
|
|
before = copy.deepcopy(initial)
|
|
|
|
response = runtime.run(initial)
|
|
|
|
self.assertTrue(response.success)
|
|
self.assertIsNotNone(response.pending_approval)
|
|
self.assertEqual(response.pending_approval.call_id, "w1") # type: ignore[union-attr]
|
|
self.assertEqual(response.tool_events[0].kind, "pending_approval")
|
|
self.assertIsNotNone(response.pending_state)
|
|
state = cast(dict[str, object], response.pending_state)
|
|
state_messages = cast(list[dict], state["messages"])
|
|
self.assertEqual(state_messages[-1], assistant)
|
|
json.dumps(response.pending_state)
|
|
self.assertEqual(initial, before)
|
|
|
|
def test_mutable_tool_requires_approval_before_executor_dispatch(self):
|
|
assistant = {
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [native_call("test-1", "run_tests", {"path": "/tmp/project"})],
|
|
}
|
|
runtime, _, executor = self.make_runtime([assistant])
|
|
|
|
response = runtime.run([{"role": "user", "content": "run the tests"}])
|
|
|
|
self.assertTrue(response.success)
|
|
self.assertEqual(response.pending_approval.name, "run_tests") # type: ignore[union-attr]
|
|
self.assertEqual(executor.calls, [])
|
|
self.assertEqual(response.tool_events[0].kind, "pending_approval")
|
|
|
|
def test_resume_processes_remaining_simultaneous_calls_in_order(self):
|
|
assistant = {
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [
|
|
native_call(
|
|
"write-1",
|
|
"write_file",
|
|
{"path": "/tmp/a", "content": "new"},
|
|
),
|
|
native_call("read-1", "read_file", {"path": "/tmp/a"}),
|
|
native_call("test-1", "run_tests", {"path": "/tmp/project"}),
|
|
native_call("list-1", "list_directory", {"path": "/tmp"}),
|
|
],
|
|
}
|
|
runtime, transport, executor = self.make_runtime([assistant, {"role": "assistant", "content": "done"}])
|
|
|
|
first = runtime.run([{"role": "user", "content": "change and inspect"}])
|
|
second = runtime.resume(
|
|
cast(dict[str, object], first.pending_state),
|
|
{"success": True, "result": "write approved"},
|
|
)
|
|
|
|
self.assertEqual(executor.calls, [("read_file", {"path": "/tmp/a"})])
|
|
self.assertEqual(second.pending_approval.name, "run_tests") # type: ignore[union-attr]
|
|
self.assertEqual(len(transport.calls), 1)
|
|
|
|
final = runtime.resume(
|
|
cast(dict[str, object], second.pending_state),
|
|
{"success": True, "result": "tests approved"},
|
|
)
|
|
|
|
self.assertTrue(final.success)
|
|
self.assertEqual(final.user_text, "done")
|
|
self.assertEqual([name for name, _ in executor.calls], ["read_file", "list_directory"])
|
|
self.assertEqual(len(transport.calls), 2)
|
|
|
|
def test_resume_preserves_consumed_iterations_across_approval(self):
|
|
responses = [
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [native_call(str(i), "list_directory", {"path": f"/tmp/{i}"})],
|
|
}
|
|
for i in range(7)
|
|
]
|
|
responses.append(
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [native_call("write-8", "write_file", {"path": "/tmp/a", "content": "x"})],
|
|
}
|
|
)
|
|
runtime, transport, _ = self.make_runtime(responses)
|
|
|
|
pending = runtime.run([{"role": "user", "content": "keep going"}])
|
|
response = runtime.resume(
|
|
cast(dict[str, object], pending.pending_state),
|
|
{"success": True, "result": "write approved"},
|
|
)
|
|
|
|
self.assertFalse(response.success)
|
|
self.assertIn("iteration limit", response.error)
|
|
self.assertEqual(len(transport.calls), 8)
|
|
|
|
def test_empty_terminal_visible_text_is_structured_failure(self):
|
|
runtime, transport, executor = self.make_runtime([{"role": "assistant", "content": " \n\t"}])
|
|
|
|
response = runtime.run([{"role": "user", "content": "hello"}])
|
|
|
|
self.assertFalse(response.success)
|
|
self.assertIn("empty", response.error)
|
|
self.assertEqual(len(transport.calls), 1)
|
|
self.assertEqual(executor.calls, [])
|
|
|
|
def test_iteration_limit_stops_after_eight_model_turns(self):
|
|
responses = [
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [native_call(str(i), "list_directory", {"path": f"/tmp/{i}"})],
|
|
}
|
|
for i in range(8)
|
|
]
|
|
runtime, transport, executor = self.make_runtime(responses)
|
|
|
|
response = runtime.run([{"role": "user", "content": "keep going"}])
|
|
|
|
self.assertFalse(response.success)
|
|
self.assertIn("iteration limit", response.error)
|
|
self.assertEqual(len(transport.calls), 8)
|
|
self.assertEqual(len(executor.calls), 8)
|
|
|
|
def test_text_fallback_is_used_only_without_native_tool_capability(self):
|
|
runtime, transport, executor = self.make_runtime(
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"content": '{"tool":"list_directory","args":{"path":"/tmp"}}',
|
|
},
|
|
{"role": "assistant", "content": "fallback done"},
|
|
],
|
|
native_tools=False,
|
|
)
|
|
|
|
response = runtime.run([{"role": "user", "content": "inspect"}])
|
|
|
|
self.assertTrue(response.success)
|
|
self.assertEqual(response.user_text, "fallback done")
|
|
self.assertEqual(len(executor.calls), 1)
|
|
self.assertIsNone(transport.calls[0]["tools"])
|
|
self.assertIn("Available Tools", transport.calls[0]["messages"][0]["content"])
|
|
|
|
def test_runtime_rejects_globally_valid_tool_outside_its_allowlist(self):
|
|
allowed = [tool for tool in TOOLS if tool["name"] == "read_file"]
|
|
runtime, _, executor = self.make_runtime(
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [
|
|
native_call(
|
|
"write-1",
|
|
"write_file",
|
|
{"path": "/tmp/a", "content": "blocked"},
|
|
)
|
|
],
|
|
}
|
|
],
|
|
tools=allowed,
|
|
)
|
|
|
|
response = runtime.run([{"role": "user", "content": "change it"}])
|
|
|
|
self.assertFalse(response.success)
|
|
self.assertIn("not available", response.error)
|
|
self.assertIsNone(response.pending_approval)
|
|
self.assertEqual(executor.calls, [])
|
|
|
|
def test_text_fallback_advertises_only_runtime_allowlist(self):
|
|
allowed = [tool for tool in TOOLS if tool["name"] == "read_file"]
|
|
runtime, transport, _ = self.make_runtime(
|
|
[{"role": "assistant", "content": "No tool needed."}],
|
|
native_tools=False,
|
|
tools=allowed,
|
|
)
|
|
|
|
response = runtime.run([{"role": "user", "content": "inspect"}])
|
|
|
|
self.assertTrue(response.success)
|
|
prompt = transport.calls[0]["messages"][0]["content"]
|
|
self.assertIn("### read_file", prompt)
|
|
self.assertNotIn("### write_file", prompt)
|
|
|
|
def test_resume_preserves_events_thinking_and_records_rejected_result(self):
|
|
assistant = {
|
|
"role": "assistant",
|
|
"content": "",
|
|
"thinking": "Inspect, then request the change.",
|
|
"tool_calls": [
|
|
native_call("read-1", "read_file", {"path": "/tmp/a"}),
|
|
native_call(
|
|
"write-1",
|
|
"write_file",
|
|
{"path": "/tmp/a", "content": "new"},
|
|
),
|
|
],
|
|
}
|
|
runtime, transport, _ = self.make_runtime(
|
|
[
|
|
assistant,
|
|
{
|
|
"role": "assistant",
|
|
"content": "I left it unchanged.",
|
|
"thinking": "The change was rejected.",
|
|
},
|
|
]
|
|
)
|
|
|
|
pending = runtime.run([{"role": "user", "content": "inspect and change"}])
|
|
response = runtime.resume(
|
|
cast(dict[str, object], pending.pending_state),
|
|
{"success": False, "error": "operator rejected"},
|
|
)
|
|
|
|
self.assertTrue(response.success)
|
|
self.assertEqual(
|
|
[event.kind for event in response.tool_events],
|
|
["completed", "pending_approval", "error"],
|
|
)
|
|
self.assertEqual(response.tool_events[-1].request.call_id, "write-1")
|
|
self.assertEqual(
|
|
response.tool_events[-1].result,
|
|
{"success": False, "error": "operator rejected"},
|
|
)
|
|
self.assertEqual(
|
|
response.thinking,
|
|
["Inspect, then request the change.", "The change was rejected."],
|
|
)
|
|
resumed_result = transport.calls[1]["messages"][-1]
|
|
self.assertEqual(resumed_result["tool_name"], "write_file")
|
|
self.assertEqual(resumed_result["tool_call_id"], "write-1")
|
|
|
|
def test_resume_rejects_non_boolean_native_tools_fail_closed(self):
|
|
runtime, transport, executor = self.make_runtime([])
|
|
pending_state: dict[str, object] = {
|
|
"messages": [{"role": "user", "content": "hello"}],
|
|
"native_tools": "false",
|
|
"iterations_used": 0,
|
|
"pending_request": {
|
|
"call_id": "write-1",
|
|
"name": "write_file",
|
|
"arguments": {"path": "/tmp/a", "content": "x"},
|
|
"source": "ollama-native",
|
|
},
|
|
}
|
|
|
|
response = runtime.resume(pending_state, {"success": True, "result": "approved"})
|
|
|
|
self.assertFalse(response.success)
|
|
self.assertIn("native tools must be a boolean", response.error)
|
|
self.assertEqual(transport.calls, [])
|
|
self.assertEqual(executor.calls, [])
|
|
|
|
def test_resume_rejects_pending_request_outside_runtime_allowlist(self):
|
|
allowed = [tool for tool in TOOLS if tool["name"] == "read_file"]
|
|
runtime, transport, executor = self.make_runtime([], tools=allowed)
|
|
pending_state: dict[str, object] = {
|
|
"messages": [{"role": "user", "content": "hello"}],
|
|
"native_tools": True,
|
|
"iterations_used": 0,
|
|
"pending_request": {
|
|
"call_id": "write-1",
|
|
"name": "write_file",
|
|
"arguments": {"path": "/tmp/a", "content": "x"},
|
|
"source": "ollama-native",
|
|
},
|
|
}
|
|
|
|
response = runtime.resume(pending_state, {"success": True, "result": "approved"})
|
|
|
|
self.assertFalse(response.success)
|
|
self.assertIn("not available", response.error)
|
|
self.assertEqual(transport.calls, [])
|
|
self.assertEqual(executor.calls, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class TestCloudModelContextDiscovery:
|
|
"""Context discovery must not hit Ollama with cloud-pinned model names.
|
|
|
|
Observed live (PL1 Task A/B, deployed 0.66.21/0.66.22): every cloud
|
|
task logged `subsystem=context_discovery tier=1 HTTPError 400` because
|
|
module-level discovery used OllamaChatTransport for a model whose
|
|
provider prefix requires the [OI]-compat wire — then misbudgeted the
|
|
task at the 8192 fallback instead of the provider's real window."""
|
|
|
|
def test_cloud_pinned_model_uses_provider_window_not_ollama(self):
|
|
import tempfile
|
|
from unittest import mock
|
|
from rune import local_tool_runtime as ltr
|
|
import rune.openai_compat_transport as _oct
|
|
ltr.reset_context_length_cache()
|
|
|
|
class _StubCloud:
|
|
def context_length(self, model):
|
|
return 131072
|
|
|
|
with mock.patch.object(ltr, "OllamaChatTransport",
|
|
side_effect=AssertionError(
|
|
"Ollama must not be consulted for cloud-pinned models")), \
|
|
mock.patch.object(_oct, "cloud_transport_for",
|
|
return_value=_StubCloud()):
|
|
window = ltr.context_length("Verboo Code/glm-5.3-flash")
|
|
assert window == 131072, window
|
|
|
|
def test_local_model_still_uses_ollama_discovery(self):
|
|
from unittest import mock
|
|
from rune import local_tool_runtime as ltr
|
|
ltr.reset_context_length_cache()
|
|
|
|
class _StubOllama:
|
|
def __init__(self, base_url="http://localhost:11434", timeout=120):
|
|
self.calls = []
|
|
|
|
def context_length(self, model):
|
|
return 4096
|
|
|
|
with mock.patch.object(ltr, "OllamaChatTransport", _StubOllama):
|
|
window = ltr.context_length("qwen2.5:3b")
|
|
assert window == 4096, window
|