wehub-resource-sync
2139 lines
71 KiB
Python
2139 lines
71 KiB
Python
"""Tests for chat command — verify CLI interface and routing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import io
|
|
import json
|
|
from collections.abc import Iterable
|
|
from contextlib import asynccontextmanager
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from rich.console import Console
|
|
from typer.testing import CliRunner
|
|
|
|
from opensquilla.cli import chat_cmd
|
|
from opensquilla.cli.main import app
|
|
from opensquilla.cli.repl import commands as repl_commands
|
|
from opensquilla.cli.repl import slash_bridge
|
|
from opensquilla.cli.repl.session_state import ChatSessionState
|
|
from opensquilla.engine.commands import DEFAULT_REGISTRY, Surface
|
|
from opensquilla.engine.types import (
|
|
ArtifactEvent,
|
|
DoneEvent,
|
|
TextDeltaEvent,
|
|
ToolResultEvent,
|
|
ToolUseStartEvent,
|
|
)
|
|
from opensquilla.session.compaction import CompactionConfig
|
|
from opensquilla.tools.types import CallerKind, ToolContext
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
def _install_fake_inputs(monkeypatch, inputs: Iterable[str]) -> None:
|
|
"""Install a fake TUI surface that yields lines from *inputs*.
|
|
|
|
The chat command reaches input through the selected TUI runtime. Patch the
|
|
TUI surface seam so these command tests stay behind the same boundary as
|
|
production code.
|
|
"""
|
|
|
|
iterator = iter(inputs)
|
|
|
|
class _FakeOutputHandle:
|
|
approval_surface = Surface.CLI_GATEWAY
|
|
|
|
async def write_through(self, payload: str) -> None:
|
|
return None
|
|
|
|
async def send_message(self, message_type: str, payload: dict[str, object]) -> None:
|
|
return None
|
|
|
|
def set_toolbar(self, key: str, value) -> None:
|
|
return None
|
|
|
|
def invalidate(self) -> None:
|
|
return None
|
|
|
|
class _FakeSurface:
|
|
output_handle = _FakeOutputHandle()
|
|
|
|
async def next_line(self) -> str | None:
|
|
try:
|
|
return next(iterator)
|
|
except StopIteration:
|
|
return None
|
|
|
|
def set_cancel_callback(self, cb) -> None:
|
|
return None
|
|
|
|
def set_shutdown_callback(self, cb) -> None:
|
|
return None
|
|
|
|
def emit_eof(self) -> None:
|
|
return None
|
|
|
|
def invalidate(self) -> None:
|
|
return None
|
|
|
|
async def write_through(self, payload: str) -> None:
|
|
return None
|
|
|
|
async def send_message(self, message_type: str, payload: dict[str, object]) -> None:
|
|
return None
|
|
|
|
@property
|
|
def redraw_callback(self):
|
|
return self.output_handle.invalidate
|
|
|
|
@asynccontextmanager
|
|
async def _fake_session(**kwargs):
|
|
yield _FakeSurface()
|
|
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.opentui.runtime.open_opentui_surface",
|
|
_fake_session,
|
|
)
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.adapters.native_bridge.open_native_terminal_surface",
|
|
_fake_session,
|
|
)
|
|
|
|
|
|
EXPECTED_GATEWAY_COMMANDS = {
|
|
"/new",
|
|
"/reset",
|
|
"/compact",
|
|
"/cmp",
|
|
"/help",
|
|
"/status",
|
|
"/model",
|
|
"/models",
|
|
"/cost",
|
|
"/usage",
|
|
"/save",
|
|
"/image",
|
|
"/path",
|
|
"/file",
|
|
"/approvals",
|
|
"/permissions",
|
|
"/forget",
|
|
"/sessions",
|
|
"/resume",
|
|
"/delete",
|
|
"/exit",
|
|
}
|
|
|
|
EXPECTED_STANDALONE_COMMANDS = EXPECTED_GATEWAY_COMMANDS - {
|
|
"/approvals",
|
|
"/delete",
|
|
"/file",
|
|
"/forget",
|
|
"/models",
|
|
"/permissions",
|
|
"/resume",
|
|
"/sessions",
|
|
"/usage",
|
|
}
|
|
|
|
|
|
def _handler_words(surface: Surface) -> set[str]:
|
|
return {word for command in DEFAULT_REGISTRY.for_surface(surface) for word in command.words()}
|
|
|
|
|
|
def test_gateway_registry_commands_have_gateway_handlers() -> None:
|
|
gateway_words = _handler_words(Surface.CLI_GATEWAY)
|
|
|
|
assert EXPECTED_GATEWAY_COMMANDS <= gateway_words
|
|
assert gateway_words == chat_cmd.GATEWAY_SLASH_HANDLER_WORDS
|
|
assert "/elevated" in chat_cmd.GATEWAY_SLASH_HANDLER_WORDS
|
|
assert "/clear" in chat_cmd.GATEWAY_SLASH_HANDLER_WORDS
|
|
assert "/quit" in chat_cmd.GATEWAY_SLASH_HANDLER_WORDS
|
|
assert "/usage" in chat_cmd.GATEWAY_SLASH_HANDLER_WORDS
|
|
assert "/file" in chat_cmd.GATEWAY_SLASH_HANDLER_WORDS
|
|
|
|
|
|
def test_standalone_registry_commands_have_standalone_handlers() -> None:
|
|
standalone_words = _handler_words(Surface.CLI_STANDALONE)
|
|
|
|
assert EXPECTED_STANDALONE_COMMANDS <= standalone_words
|
|
assert standalone_words == chat_cmd.STANDALONE_SLASH_HANDLER_WORDS
|
|
assert "/clear" in chat_cmd.STANDALONE_SLASH_HANDLER_WORDS
|
|
assert "/quit" in chat_cmd.STANDALONE_SLASH_HANDLER_WORDS
|
|
assert "/usage" not in chat_cmd.STANDALONE_SLASH_HANDLER_WORDS
|
|
assert "/file" not in chat_cmd.STANDALONE_SLASH_HANDLER_WORDS
|
|
assert "/models" not in chat_cmd.STANDALONE_SLASH_HANDLER_WORDS
|
|
|
|
|
|
def test_usage_is_gateway_only_and_not_standalone_help() -> None:
|
|
assert "/usage" in _handler_words(Surface.CLI_GATEWAY)
|
|
assert "/usage" not in _handler_words(Surface.CLI_STANDALONE)
|
|
assert "/models" in _handler_words(Surface.CLI_GATEWAY)
|
|
assert "/models" not in _handler_words(Surface.CLI_STANDALONE)
|
|
|
|
buffer = io.StringIO()
|
|
console = Console(file=buffer, force_terminal=False, width=100, highlight=False)
|
|
console.print(repl_commands.render_help_table(Surface.CLI_STANDALONE))
|
|
|
|
assert "/usage" not in buffer.getvalue()
|
|
assert "/models" not in buffer.getvalue()
|
|
|
|
|
|
def test_file_is_gateway_only() -> None:
|
|
assert "/file" in _handler_words(Surface.CLI_GATEWAY)
|
|
assert "/file" not in _handler_words(Surface.CLI_STANDALONE)
|
|
|
|
|
|
def test_interactive_chat_clear_screen_only_on_terminal(monkeypatch) -> None:
|
|
calls: list[str] = []
|
|
|
|
class FakeConsole:
|
|
is_terminal = True
|
|
|
|
def clear(self) -> None:
|
|
calls.append("clear")
|
|
|
|
monkeypatch.setattr(chat_cmd._launch_bridge, "console", FakeConsole())
|
|
|
|
chat_cmd._clear_screen_for_interactive_chat()
|
|
|
|
assert calls == ["clear"]
|
|
|
|
|
|
def test_interactive_chat_clear_screen_skips_non_terminal(monkeypatch) -> None:
|
|
calls: list[str] = []
|
|
|
|
class FakeConsole:
|
|
is_terminal = False
|
|
|
|
def clear(self) -> None:
|
|
calls.append("clear")
|
|
|
|
monkeypatch.setattr(chat_cmd._launch_bridge, "console", FakeConsole())
|
|
|
|
chat_cmd._clear_screen_for_interactive_chat()
|
|
|
|
assert calls == []
|
|
|
|
|
|
class TestChatCommand:
|
|
def test_chat_command_builds_typed_launch_request(
|
|
self,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_launch_chat_command(request, *, overrides=None) -> None:
|
|
captured["request"] = request
|
|
captured["overrides"] = overrides
|
|
|
|
monkeypatch.setattr(
|
|
chat_cmd,
|
|
"_launch_chat_command",
|
|
fake_launch_chat_command,
|
|
raising=False,
|
|
)
|
|
|
|
chat_cmd.run_chat(
|
|
model="openrouter/test",
|
|
session_id="agent:main:existing",
|
|
standalone=True,
|
|
workspace="repo",
|
|
workspace_strict=True,
|
|
timeout=12.5,
|
|
)
|
|
|
|
request = captured["request"]
|
|
assert isinstance(request, chat_cmd._ChatCommandRequest)
|
|
assert request.model == "openrouter/test"
|
|
assert request.session_id == "agent:main:existing"
|
|
assert request.standalone is True
|
|
assert request.workspace == "repo"
|
|
assert request.workspace_strict is True
|
|
assert request.timeout == 12.5
|
|
overrides = captured["overrides"]
|
|
assert isinstance(overrides, chat_cmd._ChatCommandLaunchOverrides)
|
|
assert overrides.launch_chat is None
|
|
assert overrides.standalone_runner is None
|
|
assert overrides.gateway_runner is None
|
|
|
|
def test_chat_help(self) -> None:
|
|
result = runner.invoke(
|
|
app,
|
|
["chat", "--help"],
|
|
env={"COLUMNS": "120", "NO_COLOR": "1", "TERM": "dumb"},
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "--model" in result.output
|
|
assert "--session" in result.output
|
|
|
|
def test_chat_invokes_run_chat(self) -> None:
|
|
"""Default chat calls run_chat with correct defaults."""
|
|
mock_run = MagicMock()
|
|
with patch("opensquilla.cli.chat_cmd.run_chat", mock_run):
|
|
result = runner.invoke(app, ["chat"])
|
|
assert result.exit_code == 0
|
|
mock_run.assert_called_once_with(
|
|
model="",
|
|
session_id="",
|
|
standalone=False,
|
|
workspace="",
|
|
workspace_strict=None,
|
|
timeout=None,
|
|
)
|
|
|
|
def test_chat_model_option_forwarded(self) -> None:
|
|
"""--model option is forwarded to run_chat."""
|
|
mock_run = MagicMock()
|
|
with patch("opensquilla.cli.chat_cmd.run_chat", mock_run):
|
|
result = runner.invoke(app, ["chat", "--model", "ollama/llama3"])
|
|
assert result.exit_code == 0
|
|
mock_run.assert_called_once_with(
|
|
model="ollama/llama3",
|
|
session_id="",
|
|
standalone=False,
|
|
workspace="",
|
|
workspace_strict=None,
|
|
timeout=None,
|
|
)
|
|
|
|
def test_chat_session_option_forwarded(self) -> None:
|
|
"""--session option is forwarded to run_chat."""
|
|
mock_run = MagicMock()
|
|
with patch("opensquilla.cli.chat_cmd.run_chat", mock_run):
|
|
result = runner.invoke(app, ["chat", "--session", "abc123"])
|
|
assert result.exit_code == 0
|
|
mock_run.assert_called_once_with(
|
|
model="",
|
|
session_id="abc123",
|
|
standalone=False,
|
|
workspace="",
|
|
workspace_strict=None,
|
|
timeout=None,
|
|
)
|
|
|
|
def test_chat_rejects_unknown_tui_backend_before_launch(self) -> None:
|
|
result = runner.invoke(
|
|
app,
|
|
["chat"],
|
|
env={
|
|
"OPENSQUILLA_TUI_BACKEND": "bogus",
|
|
"COLUMNS": "120",
|
|
"NO_COLOR": "1",
|
|
"TERM": "dumb",
|
|
},
|
|
)
|
|
|
|
assert result.exit_code == 2
|
|
assert "Unsupported TUI backend" in result.output
|
|
assert "bogus" in result.output
|
|
assert "opentui" in result.output
|
|
|
|
def test_chat_timeout_option_forwarded(self) -> None:
|
|
"""--timeout option is forwarded to run_chat."""
|
|
mock_run = MagicMock()
|
|
with patch("opensquilla.cli.chat_cmd.run_chat", mock_run):
|
|
result = runner.invoke(app, ["chat", "--timeout", "12.5"])
|
|
assert result.exit_code == 0
|
|
mock_run.assert_called_once_with(
|
|
model="",
|
|
session_id="",
|
|
standalone=False,
|
|
workspace="",
|
|
workspace_strict=None,
|
|
timeout=12.5,
|
|
)
|
|
|
|
def test_chat_workspace_options_forwarded(self) -> None:
|
|
mock_run = MagicMock()
|
|
with patch("opensquilla.cli.chat_cmd.run_chat", mock_run):
|
|
result = runner.invoke(
|
|
app,
|
|
["chat", "--workspace", "repo", "--workspace-strict"],
|
|
)
|
|
assert result.exit_code == 0
|
|
mock_run.assert_called_once_with(
|
|
model="",
|
|
session_id="",
|
|
standalone=False,
|
|
workspace="repo",
|
|
workspace_strict=True,
|
|
timeout=None,
|
|
)
|
|
|
|
def test_gateway_chat_workspace_options_warn_without_forwarding(
|
|
self,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
buffer = io.StringIO()
|
|
called: dict[str, object] = {}
|
|
|
|
async def fake_gateway_chat(model: str | None, session_id: str | None) -> None:
|
|
called["model"] = model
|
|
called["session_id"] = session_id
|
|
|
|
monkeypatch.setattr(
|
|
chat_cmd._launch_bridge.sys,
|
|
"stdin",
|
|
SimpleNamespace(isatty=lambda: True),
|
|
)
|
|
# The real log-quieting swaps the process-global structlog config to an
|
|
# interactive log file; leaking that into later tests poisons every
|
|
# subsequent structlog emit once another test closes the handle.
|
|
monkeypatch.setattr(
|
|
chat_cmd._launch_bridge,
|
|
"quiet_logs_for_interactive_chat",
|
|
lambda: None,
|
|
)
|
|
monkeypatch.setattr(
|
|
chat_cmd._launch_bridge,
|
|
"console",
|
|
Console(file=buffer, force_terminal=True, color_system=None, no_color=True),
|
|
)
|
|
monkeypatch.setattr(chat_cmd, "_gateway_chat", fake_gateway_chat)
|
|
|
|
chat_cmd.run_chat(
|
|
model="",
|
|
session_id="",
|
|
standalone=False,
|
|
workspace="repo",
|
|
workspace_strict=True,
|
|
timeout=None,
|
|
)
|
|
|
|
assert called == {"model": None, "session_id": None}
|
|
output = buffer.getvalue()
|
|
assert "--workspace only affects --standalone chat" in output
|
|
assert "requires the path to be visible to the gateway runtime" in output
|
|
|
|
def test_chat_rejects_extra_args(self) -> None:
|
|
"""Extra positional args (like 'send Hello') are rejected."""
|
|
result = runner.invoke(app, ["chat", "send", "Hello"])
|
|
assert result.exit_code != 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_repl_delegates_to_runtime_bridge(monkeypatch) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
async def fake_run_standalone_chat(**kwargs) -> None:
|
|
captured.update(kwargs)
|
|
|
|
monkeypatch.setattr(
|
|
chat_cmd._runtime_bridge,
|
|
"run_standalone_chat",
|
|
fake_run_standalone_chat,
|
|
)
|
|
|
|
await chat_cmd._standalone_repl(
|
|
model="openrouter/test",
|
|
session_id="standalone:test",
|
|
workspace="repo",
|
|
workspace_strict=True,
|
|
timeout=7.25,
|
|
)
|
|
|
|
assert captured["model"] == "openrouter/test"
|
|
assert captured["session_id"] == "standalone:test"
|
|
assert captured["workspace"] == "repo"
|
|
assert captured["workspace_strict"] is True
|
|
assert captured["timeout"] == 7.25
|
|
assert "stream_response" not in captured
|
|
assert "image_command_handler" not in captured
|
|
assert "run_concurrent_repl" not in captured
|
|
assert "output_console" not in captured
|
|
assert "error_panel_factory" not in captured
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_chat_delegates_to_runtime_bridge(monkeypatch) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
async def fake_run_gateway_chat(**kwargs) -> None:
|
|
captured.update(kwargs)
|
|
|
|
monkeypatch.setattr(
|
|
chat_cmd._runtime_bridge,
|
|
"run_gateway_chat",
|
|
fake_run_gateway_chat,
|
|
)
|
|
|
|
await chat_cmd._gateway_chat(
|
|
model="anthropic/claude-sonnet-4",
|
|
session_id="agent:main:resumed-key",
|
|
)
|
|
|
|
assert captured["model"] == "anthropic/claude-sonnet-4"
|
|
assert captured["session_id"] == "agent:main:resumed-key"
|
|
assert "stream_response" not in captured
|
|
assert "handle_slash_command" not in captured
|
|
assert "run_concurrent_repl" not in captured
|
|
assert "output_console" not in captured
|
|
assert "error_panel_factory" not in captured
|
|
|
|
|
|
class _FakeSessionManager:
|
|
def __init__(self) -> None:
|
|
self.get_or_create_calls: list[dict[str, str]] = []
|
|
self.compact_calls: list[tuple[str, int, object | None]] = []
|
|
self.truncate_calls: list[tuple[str, int]] = []
|
|
self.transcripts: dict[str, list[object]] = {}
|
|
|
|
async def get_or_create(self, session_key: str, agent_id: str = "main") -> object:
|
|
self.get_or_create_calls.append({"session_key": session_key, "agent_id": agent_id})
|
|
return SimpleNamespace(session_key=session_key, agent_id=agent_id)
|
|
|
|
async def append_message(self, session_key: str, role: str, content: str) -> object:
|
|
entry = SimpleNamespace(role=role, content=content)
|
|
self.transcripts.setdefault(session_key, []).append(entry)
|
|
return entry
|
|
|
|
async def get_transcript(self, session_key: str) -> list[object]:
|
|
return list(self.transcripts.get(session_key, []))
|
|
|
|
async def truncate(self, session_key: str, max_messages: int = 0) -> None:
|
|
self.truncate_calls.append((session_key, max_messages))
|
|
if max_messages <= 0:
|
|
self.transcripts[session_key] = []
|
|
else:
|
|
self.transcripts[session_key] = self.transcripts.get(session_key, [])[-max_messages:]
|
|
|
|
async def compact(self, session_key: str, context_window_tokens: int, config=None) -> str:
|
|
self.compact_calls.append((session_key, context_window_tokens, config))
|
|
return "summary"
|
|
|
|
|
|
class _LegacyCompactSessionManager(_FakeSessionManager):
|
|
async def compact(self, session_key: str, context_window_tokens: int) -> str:
|
|
self.compact_calls.append((session_key, context_window_tokens, None))
|
|
return "summary"
|
|
|
|
|
|
class _FakeCompactionProvider:
|
|
provider_name = "openai"
|
|
|
|
def __init__(self) -> None:
|
|
self._api_key = "cli-provider-key"
|
|
self._model = "provider/model"
|
|
self._base_url = "https://openrouter.ai/api/v1"
|
|
|
|
@property
|
|
def model(self) -> str:
|
|
return self._model
|
|
|
|
|
|
class _FakeProviderSelector:
|
|
def __init__(self, provider: _FakeCompactionProvider | None = None) -> None:
|
|
self.provider = provider or _FakeCompactionProvider()
|
|
|
|
def clone(self) -> _FakeProviderSelector:
|
|
return self
|
|
|
|
def resolve(self) -> _FakeCompactionProvider:
|
|
return self.provider
|
|
|
|
|
|
class _FakeServices:
|
|
def __init__(self) -> None:
|
|
self.memory_sync_managers = {"main": object()}
|
|
self.memory_retrievers = {"main": object()}
|
|
self.turn_capture_services = {"main": object()}
|
|
self.flush_service = None
|
|
self.model_catalog = object()
|
|
self.provider_selector = MagicMock()
|
|
self.tool_registry = None
|
|
self.session_manager = _FakeSessionManager()
|
|
self.skill_loader = None
|
|
self.usage_tracker = None
|
|
self.config = None
|
|
|
|
async def close(self) -> None:
|
|
return None
|
|
|
|
|
|
class _DummyLive:
|
|
def __init__(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
def __enter__(self) -> _DummyLive:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> bool:
|
|
return False
|
|
|
|
def update(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
|
|
class _RecordingRenderer:
|
|
instances: list[_RecordingRenderer] = []
|
|
|
|
def __init__(self, *args, **kwargs) -> None:
|
|
self.buffer = ""
|
|
self.pulses = 0
|
|
self.errors: list[str] = []
|
|
self.finalized = False
|
|
self.statuses: list[str] = []
|
|
self.tool_starts: list[tuple[str, object, str | None]] = []
|
|
self.tool_finishes: list[tuple[str | None, bool, object | None]] = []
|
|
_RecordingRenderer.instances.append(self)
|
|
|
|
def __enter__(self) -> _RecordingRenderer:
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> bool:
|
|
return False
|
|
|
|
def append_text(self, delta: str) -> None:
|
|
self.buffer += delta
|
|
|
|
async def aappend_text(self, delta: str) -> None:
|
|
# Async sibling used by production stream paths; mirror sync logic
|
|
# so existing test assertions against `self.buffer` still hold.
|
|
self.buffer += delta
|
|
|
|
def pulse(self) -> None:
|
|
self.pulses += 1
|
|
|
|
def tool_call(self, name: str, args=None) -> None:
|
|
# Legacy shim kept for tests that haven't migrated.
|
|
self.tool_starts.append((name, args, None))
|
|
|
|
def tool_start(self, name: str, args=None, tool_use_id: str | None = None) -> None:
|
|
self.tool_starts.append((name, args, tool_use_id))
|
|
|
|
def tool_finished(
|
|
self,
|
|
tool_use_id: str | None,
|
|
*,
|
|
success: bool,
|
|
elapsed: float | None = None,
|
|
error: str | None = None,
|
|
result: object | None = None,
|
|
) -> None:
|
|
self.tool_finishes.append((tool_use_id, success, result))
|
|
|
|
def status(self, message: str, **_kwargs) -> None:
|
|
self.statuses.append(message)
|
|
|
|
def error(self, message: str) -> None:
|
|
self.errors.append(message)
|
|
|
|
def finalize(
|
|
self,
|
|
usage=None,
|
|
*,
|
|
cancelled: bool = False,
|
|
) -> None:
|
|
self.finalized = True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_repl_forwards_timeout(monkeypatch) -> None:
|
|
captured: dict[str, object] = {}
|
|
inputs = iter(["hello", "/quit"])
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
captured["message"] = message
|
|
captured["session_key"] = session_key
|
|
captured["timeout"] = kwargs.get("timeout")
|
|
captured["tool_context"] = kwargs["tool_context"]
|
|
yield DoneEvent()
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return _FakeServices()
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._standalone_repl(
|
|
model="openrouter/test",
|
|
session_id="standalone:test",
|
|
timeout=7.25,
|
|
)
|
|
|
|
assert captured["message"] == "hello"
|
|
assert captured["session_key"] == "standalone:test"
|
|
assert captured["timeout"] == 7.25
|
|
assert captured["tool_context"].channel_kind == "cli"
|
|
assert captured["tool_context"].channel_id == "cli:chat"
|
|
assert captured["tool_context"].sender_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_chat_uses_workspace_in_tool_context(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
captured: dict[str, object] = {}
|
|
inputs = iter(["hello", "/quit"])
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
captured["tool_context"] = kwargs["tool_context"]
|
|
yield DoneEvent()
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return _FakeServices()
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._standalone_repl(
|
|
model="openrouter/test",
|
|
session_id="standalone:test",
|
|
workspace=str(tmp_path),
|
|
workspace_strict=True,
|
|
)
|
|
|
|
tool_context = captured["tool_context"]
|
|
assert tool_context.workspace_dir == str(tmp_path)
|
|
assert tool_context.workspace_strict is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_path_command_runs_as_plain_message(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
target = tmp_path / "large.log"
|
|
target.write_text("hello\n", encoding="utf-8")
|
|
captured: dict[str, object] = {}
|
|
inputs = iter([f"/path {target} inspect", "/quit"])
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
captured["message"] = message
|
|
captured["kwargs"] = kwargs
|
|
yield DoneEvent()
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return _FakeServices()
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._standalone_repl(
|
|
model="openrouter/test",
|
|
session_id="standalone:test",
|
|
)
|
|
|
|
assert "inspect" in captured["message"]
|
|
assert str(target.resolve(strict=False)) in captured["message"]
|
|
assert "attachments" not in captured["kwargs"]
|
|
|
|
|
|
def test_chat_workspace_strict_resolution_matches_agent_precedence(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
from opensquilla.cli.agent_cmd import _resolve_workspace_strict
|
|
|
|
monkeypatch.setenv("OPENSQUILLA_WORKSPACE_STRICT", "false")
|
|
assert (
|
|
_resolve_workspace_strict(
|
|
cli_value=True,
|
|
config_value=False,
|
|
entrypoint_default=bool(tmp_path),
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
_resolve_workspace_strict(
|
|
cli_value=None,
|
|
config_value=True,
|
|
entrypoint_default=bool(tmp_path),
|
|
)
|
|
is False
|
|
)
|
|
monkeypatch.delenv("OPENSQUILLA_WORKSPACE_STRICT")
|
|
assert (
|
|
_resolve_workspace_strict(
|
|
cli_value=None,
|
|
config_value=True,
|
|
entrypoint_default=False,
|
|
)
|
|
is True
|
|
)
|
|
assert (
|
|
_resolve_workspace_strict(
|
|
cli_value=None,
|
|
config_value=None,
|
|
entrypoint_default=True,
|
|
)
|
|
is True
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_repl_wires_memory_services_into_turnrunner(monkeypatch) -> None:
|
|
services = _FakeServices()
|
|
captured: dict[str, object] = {}
|
|
inputs = iter(["/quit"])
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs) -> None:
|
|
captured.update(kwargs)
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
yield DoneEvent()
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return services
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._standalone_repl(
|
|
model="openrouter/test",
|
|
session_id="standalone:test",
|
|
timeout=7.25,
|
|
)
|
|
|
|
assert captured["memory_sync_managers"] is services.memory_sync_managers
|
|
assert captured["memory_retrievers"] is services.memory_retrievers
|
|
assert captured["turn_capture_services"] is services.turn_capture_services
|
|
assert captured["session_flush_service"] is services.flush_service
|
|
assert captured["model_catalog"] is services.model_catalog
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_turnrunner_stream_uses_heartbeat_wrapper(monkeypatch) -> None:
|
|
class FakeTurnRunner:
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
await asyncio.sleep(0.03)
|
|
yield TextDeltaEvent(text="ok")
|
|
yield DoneEvent()
|
|
|
|
_RecordingRenderer.instances.clear()
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.opentui.renderer.OpenTuiStreamRenderer",
|
|
_RecordingRenderer,
|
|
)
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.adapters.turn_stream_defaults.OpenTuiStreamRenderer",
|
|
_RecordingRenderer,
|
|
)
|
|
# Renderer choice is now backend-aware; pin the OpenTUI backend so the
|
|
# recording renderer (patched above) is the one the bridge selects.
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.adapters.runtime_bridge.validate_tui_backend_selection",
|
|
lambda env=None: "opentui",
|
|
)
|
|
svc = SimpleNamespace(
|
|
config=SimpleNamespace(
|
|
agent_stream_heartbeat_interval_seconds=0.01,
|
|
agent_stream_idle_timeout_seconds=1.0,
|
|
),
|
|
session_manager=_FakeSessionManager(),
|
|
)
|
|
tool_ctx = ToolContext(caller_kind=CallerKind.CLI, channel_kind="cli", channel_id="cli:chat")
|
|
|
|
result = await chat_cmd._stream_response_turnrunner(
|
|
FakeTurnRunner(),
|
|
"agent:main:standalone:test",
|
|
tool_ctx,
|
|
"hello",
|
|
svc=svc,
|
|
)
|
|
|
|
renderer = _RecordingRenderer.instances[-1]
|
|
assert result.text == "ok"
|
|
assert renderer.pulses >= 1
|
|
assert renderer.finalized is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_turnrunner_stream_collects_artifacts(monkeypatch) -> None:
|
|
artifact = {
|
|
"id": "art-chat",
|
|
"kind": "artifact_ref",
|
|
"name": "report.txt",
|
|
"mime": "text/plain",
|
|
"size": 4,
|
|
"sha256": "e" * 64,
|
|
"session_id": "session-1",
|
|
"session_key": "agent:main:standalone:test",
|
|
"source": "publish_artifact",
|
|
"created_at": "2026-05-06T12:00:00Z",
|
|
"download_url": "/api/v1/artifacts/art-chat?sessionKey=agent%3Amain%3Astandalone%3Atest",
|
|
"store": "artifacts",
|
|
}
|
|
|
|
class FakeTurnRunner:
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
yield ArtifactEvent(**artifact)
|
|
yield TextDeltaEvent(text="ok")
|
|
yield DoneEvent()
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.opentui.renderer.OpenTuiStreamRenderer",
|
|
_RecordingRenderer,
|
|
)
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.adapters.turn_stream_defaults.OpenTuiStreamRenderer",
|
|
_RecordingRenderer,
|
|
)
|
|
# Renderer choice is now backend-aware; pin the OpenTUI backend so the
|
|
# recording renderer (patched above) is the one the bridge selects.
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.adapters.runtime_bridge.validate_tui_backend_selection",
|
|
lambda env=None: "opentui",
|
|
)
|
|
svc = SimpleNamespace(
|
|
config=SimpleNamespace(
|
|
agent_stream_heartbeat_interval_seconds=0.0,
|
|
agent_stream_idle_timeout_seconds=1.0,
|
|
),
|
|
session_manager=_FakeSessionManager(),
|
|
)
|
|
tool_ctx = ToolContext(caller_kind=CallerKind.CLI, channel_kind="cli", channel_id="cli:chat")
|
|
|
|
result = await chat_cmd._stream_response_turnrunner(
|
|
FakeTurnRunner(),
|
|
"agent:main:standalone:test",
|
|
tool_ctx,
|
|
"hello",
|
|
svc=svc,
|
|
)
|
|
|
|
assert result.text == "ok"
|
|
assert result.artifacts[0]["download_url"] == "/api/v1/artifacts/art-chat"
|
|
assert "session_key" not in result.artifacts[0]
|
|
assert "sessionKey" not in json.dumps(result.artifacts[0])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_turnrunner_wires_tool_strip(monkeypatch) -> None:
|
|
"""Tool events must drive ``tool_start``/``tool_finished`` on the renderer.
|
|
|
|
Without this wiring the ToolCallStrip pairing/coalescing never runs against
|
|
real traffic and the 12-call ``execute_code`` flood reappears.
|
|
"""
|
|
|
|
class FakeTurnRunner:
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
yield ToolUseStartEvent(tool_use_id="call-1", tool_name="execute_code")
|
|
yield ToolResultEvent(
|
|
tool_use_id="call-1",
|
|
tool_name="execute_code",
|
|
result="2",
|
|
is_error=False,
|
|
)
|
|
yield ToolUseStartEvent(tool_use_id="call-2", tool_name="execute_code")
|
|
yield ToolResultEvent(
|
|
tool_use_id="call-2",
|
|
tool_name="execute_code",
|
|
result="boom",
|
|
is_error=True,
|
|
)
|
|
yield ToolUseStartEvent(tool_use_id="call-3", tool_name="execute_code")
|
|
yield ToolResultEvent(
|
|
tool_use_id="call-3",
|
|
tool_name="execute_code",
|
|
result="approval pending",
|
|
is_error=False,
|
|
execution_status={
|
|
"version": 1,
|
|
"status": "unknown",
|
|
"exit_code": None,
|
|
"timed_out": False,
|
|
"truncated": False,
|
|
"reason": "approval_pending",
|
|
"source": "tool_runtime",
|
|
"preservation_class": "ephemeral",
|
|
},
|
|
)
|
|
yield DoneEvent()
|
|
|
|
_RecordingRenderer.instances.clear()
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.opentui.renderer.OpenTuiStreamRenderer",
|
|
_RecordingRenderer,
|
|
)
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.adapters.turn_stream_defaults.OpenTuiStreamRenderer",
|
|
_RecordingRenderer,
|
|
)
|
|
# Renderer choice is now backend-aware; pin the OpenTUI backend so the
|
|
# recording renderer (patched above) is the one the bridge selects.
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.adapters.runtime_bridge.validate_tui_backend_selection",
|
|
lambda env=None: "opentui",
|
|
)
|
|
svc = SimpleNamespace(
|
|
config=SimpleNamespace(
|
|
agent_stream_heartbeat_interval_seconds=0.0,
|
|
agent_stream_idle_timeout_seconds=1.0,
|
|
),
|
|
session_manager=_FakeSessionManager(),
|
|
)
|
|
tool_ctx = ToolContext(caller_kind=CallerKind.CLI, channel_kind="cli", channel_id="cli:chat")
|
|
|
|
await chat_cmd._stream_response_turnrunner(
|
|
FakeTurnRunner(),
|
|
"agent:main:standalone:test",
|
|
tool_ctx,
|
|
"hello",
|
|
svc=svc,
|
|
)
|
|
|
|
renderer = _RecordingRenderer.instances[-1]
|
|
assert renderer.tool_starts == [
|
|
("execute_code", None, "call-1"),
|
|
("execute_code", None, "call-2"),
|
|
("execute_code", None, "call-3"),
|
|
]
|
|
assert renderer.tool_finishes == [
|
|
("call-1", True, "2"),
|
|
("call-2", False, "boom"),
|
|
("call-3", False, "approval pending"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_repl_uses_exact_slash_tokens(monkeypatch) -> None:
|
|
services = _FakeServices()
|
|
inputs = iter(["/newer", "/models", "/quit"])
|
|
run_calls: list[str] = []
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
run_calls.append(message)
|
|
yield DoneEvent()
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return services
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._standalone_repl(
|
|
model="openrouter/test",
|
|
session_id="standalone:test",
|
|
timeout=7.25,
|
|
)
|
|
|
|
assert services.session_manager.get_or_create_calls == [
|
|
{"session_key": "standalone:test", "agent_id": "main"}
|
|
]
|
|
assert run_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_slash_compact_passes_provider_config(monkeypatch) -> None:
|
|
services = _FakeServices()
|
|
services.provider_selector = _FakeProviderSelector()
|
|
services.config = SimpleNamespace(
|
|
context_budget_tokens=1234,
|
|
compaction=SimpleNamespace(enabled=True, model=None, timeout_seconds=12.5),
|
|
)
|
|
inputs = iter(["/compact", "/quit"])
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
yield DoneEvent()
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return services
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._standalone_repl(
|
|
model="openrouter/test",
|
|
session_id="standalone:test",
|
|
timeout=7.25,
|
|
)
|
|
|
|
assert len(services.session_manager.compact_calls) == 1
|
|
session_key, context_window, config = services.session_manager.compact_calls[0]
|
|
assert session_key == "standalone:test"
|
|
assert context_window == 1234
|
|
assert isinstance(config, CompactionConfig)
|
|
assert config.api_key == "cli-provider-key"
|
|
assert config.model == "openrouter/test"
|
|
assert config.base_url == "https://openrouter.ai/api/v1"
|
|
assert config.timeout_seconds == 12.5
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_reset_refuses_non_empty_transcript_without_flush_service(
|
|
monkeypatch,
|
|
) -> None:
|
|
services = _FakeServices()
|
|
services.flush_service = None
|
|
session_key = "standalone:test"
|
|
services.session_manager.transcripts[session_key] = [
|
|
SimpleNamespace(role="user", content="persisted")
|
|
]
|
|
inputs = iter(["/reset", "/quit"])
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
yield DoneEvent()
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return services
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._standalone_repl(model="openrouter/test", session_id=session_key)
|
|
|
|
assert services.session_manager.truncate_calls == []
|
|
assert await services.session_manager.get_transcript(session_key)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_compact_missing_flush_service_does_not_block_compaction(
|
|
monkeypatch,
|
|
) -> None:
|
|
services = _FakeServices()
|
|
services.flush_service = None
|
|
session_key = "standalone:test"
|
|
services.session_manager.transcripts[session_key] = [
|
|
SimpleNamespace(role="user", content="persisted")
|
|
]
|
|
services.provider_selector = _FakeProviderSelector()
|
|
services.config = SimpleNamespace(
|
|
context_budget_tokens=1234,
|
|
compaction=SimpleNamespace(enabled=True, model=None, timeout_seconds=12.5),
|
|
)
|
|
inputs = iter(["/compact", "/quit"])
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
yield DoneEvent()
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return services
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._standalone_repl(model="openrouter/test", session_id=session_key)
|
|
|
|
assert len(services.session_manager.compact_calls) == 1
|
|
|
|
|
|
class _FakeFlushService:
|
|
def __init__(self, receipt: object | None = None, error: Exception | None = None) -> None:
|
|
self.receipt = receipt or SimpleNamespace(
|
|
mode="llm",
|
|
error=None,
|
|
indexed_chunk_count=1,
|
|
integrity_status="ok",
|
|
output_coverage_status="ok",
|
|
invalid_candidate_count=0,
|
|
candidate_missing_ids=[],
|
|
obligation_status="ok",
|
|
obligation_missing_ids=[],
|
|
)
|
|
self.error = error
|
|
self.calls: list[dict[str, object]] = []
|
|
|
|
async def execute(self, transcript: object, session_key: str, **kwargs) -> object:
|
|
self.calls.append({"transcript": transcript, "session_key": session_key, "kwargs": kwargs})
|
|
if self.error is not None:
|
|
raise self.error
|
|
return self.receipt
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_compact_flushes_before_compacting(monkeypatch) -> None:
|
|
services = _FakeServices()
|
|
session_key = "standalone:test"
|
|
services.session_manager.transcripts[session_key] = [
|
|
SimpleNamespace(role="user", content="persisted")
|
|
]
|
|
services.flush_service = _FakeFlushService()
|
|
services.provider_selector = _FakeProviderSelector()
|
|
services.config = SimpleNamespace(
|
|
context_budget_tokens=1234,
|
|
compaction=SimpleNamespace(enabled=True, model=None, timeout_seconds=12.5),
|
|
)
|
|
inputs = iter(["/compact", "/quit"])
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
yield DoneEvent()
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return services
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._standalone_repl(model="openrouter/test", session_id=session_key)
|
|
|
|
assert len(services.flush_service.calls) == 1
|
|
assert services.flush_service.calls[0]["session_key"] == session_key
|
|
assert services.flush_service.calls[0]["kwargs"]["message_window"] == 0
|
|
assert services.flush_service.calls[0]["kwargs"]["segment_mode"] == "auto"
|
|
assert len(services.session_manager.compact_calls) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_compact_continues_when_flush_fails(monkeypatch) -> None:
|
|
services = _FakeServices()
|
|
session_key = "standalone:test"
|
|
services.session_manager.transcripts[session_key] = [
|
|
SimpleNamespace(role="user", content="persisted")
|
|
]
|
|
services.flush_service = _FakeFlushService(
|
|
receipt=SimpleNamespace(mode="error", error="provider down")
|
|
)
|
|
services.provider_selector = _FakeProviderSelector()
|
|
services.config = SimpleNamespace(
|
|
context_budget_tokens=1234,
|
|
compaction=SimpleNamespace(enabled=True, model=None, timeout_seconds=12.5),
|
|
)
|
|
inputs = iter(["/compact", "/quit"])
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
yield DoneEvent()
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return services
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._standalone_repl(model="openrouter/test", session_id=session_key)
|
|
|
|
assert len(services.flush_service.calls) == 1
|
|
assert len(services.session_manager.compact_calls) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_slash_compact_keeps_legacy_compact_manager_compatible(
|
|
monkeypatch,
|
|
) -> None:
|
|
services = _FakeServices()
|
|
services.session_manager = _LegacyCompactSessionManager()
|
|
services.provider_selector = _FakeProviderSelector()
|
|
services.config = SimpleNamespace(
|
|
context_budget_tokens=1234,
|
|
compaction=SimpleNamespace(enabled=True, model=None, timeout_seconds=12.5),
|
|
)
|
|
inputs = iter(["/compact", "/quit"])
|
|
|
|
class FakeTurnRunner:
|
|
def __init__(self, **kwargs) -> None:
|
|
return None
|
|
|
|
async def run(self, message: str, session_key: str, **kwargs):
|
|
yield DoneEvent()
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return services
|
|
|
|
monkeypatch.setattr("opensquilla.engine.runtime.TurnRunner", FakeTurnRunner)
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._standalone_repl(
|
|
model="openrouter/test",
|
|
session_id="standalone:test",
|
|
timeout=7.25,
|
|
)
|
|
|
|
assert services.session_manager.compact_calls == [("standalone:test", 1234, None)]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gateway-mode flag forwarding
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakeGatewayClient:
|
|
"""Fake GatewayClient that records create/send calls and feeds the REPL exit.
|
|
|
|
Patched in place of the real `GatewayClient` class so `_stream_response_gateway`'s
|
|
``isinstance(client, GatewayClient)`` assertion passes. Each instance registers
|
|
itself in a class-level ``instances`` list so tests can grab the one created
|
|
by the function-under-test.
|
|
"""
|
|
|
|
instances: list[_FakeGatewayClient]
|
|
|
|
def __init__(self) -> None:
|
|
self.create_calls: list[dict[str, object]] = []
|
|
self.send_calls: list[dict[str, object]] = []
|
|
self.resolve_calls: list[str] = []
|
|
self.delete_calls: list[list[str]] = []
|
|
self.history_calls: list[dict[str, object]] = []
|
|
self.abort_calls: list[str] = []
|
|
self.reset_calls: list[str] = []
|
|
self.compact_calls: list[dict[str, object]] = []
|
|
self.config_get_calls: list[str | None] = []
|
|
self.config_patch_safe_calls: list[dict[str, object]] = []
|
|
self.config_values: dict[str, object] = {}
|
|
self.usage_status_calls = 0
|
|
self.list_models_calls = 0
|
|
self.delete_result: dict[str, object] = {"deleted": [], "errors": []}
|
|
self.resolved_payload: dict[str, object] = {
|
|
"session_key": "agent:main:resolved",
|
|
"model": "openai/test",
|
|
}
|
|
self.connected = False
|
|
self.closed = False
|
|
type(self).instances.append(self)
|
|
|
|
async def connect(self, url: str, *, token: str | None = None) -> None:
|
|
self.connected = True
|
|
|
|
async def create_session(
|
|
self,
|
|
agent_id: str = "main",
|
|
model: str | None = None,
|
|
display_name: str | None = None,
|
|
) -> str:
|
|
self.create_calls.append(
|
|
{"agent_id": agent_id, "model": model, "display_name": display_name}
|
|
)
|
|
return "agent:main:fake12345"
|
|
|
|
async def resolve_session(self, key: str) -> dict[str, object]:
|
|
self.resolve_calls.append(key)
|
|
return self.resolved_payload
|
|
|
|
async def delete_sessions(self, keys: list[str]) -> dict[str, object]:
|
|
self.delete_calls.append(keys)
|
|
return self.delete_result
|
|
|
|
async def session_history(self, session_key: str, limit: int = 1000) -> dict[str, object]:
|
|
self.history_calls.append({"session_key": session_key, "limit": limit})
|
|
return {
|
|
"messages": [
|
|
{"role": "user", "text": "persisted hello"},
|
|
{"role": "assistant", "text": "persisted reply"},
|
|
]
|
|
}
|
|
|
|
async def list_models(self) -> list[dict[str, object]]:
|
|
self.list_models_calls += 1
|
|
return [{"id": "openai/test", "provider": "openai"}]
|
|
|
|
async def usage_status(self) -> dict[str, object]:
|
|
self.usage_status_calls += 1
|
|
return {"totalTokens": 1234, "totalCostUsd": 0.05678}
|
|
|
|
async def abort_session(self, session_key: str) -> dict[str, object]:
|
|
self.abort_calls.append(session_key)
|
|
return {"aborted": True, "key": session_key}
|
|
|
|
async def reset_session(self, session_key: str) -> dict[str, object]:
|
|
self.reset_calls.append(session_key)
|
|
return {"reset": True, "key": session_key}
|
|
|
|
async def compact_session(self, session_key: str) -> dict[str, object]:
|
|
self.compact_calls.append({"session_key": session_key})
|
|
return {
|
|
"key": session_key,
|
|
"compacted": True,
|
|
"mode": "summary",
|
|
"summary_len": 37,
|
|
}
|
|
|
|
async def get_config(self, path: str | None = None) -> object:
|
|
self.config_get_calls.append(path)
|
|
if path is None:
|
|
return dict(self.config_values)
|
|
return self.config_values.get(path)
|
|
|
|
async def patch_config_safe(self, patches: dict[str, object]) -> dict[str, object]:
|
|
self.config_patch_safe_calls.append(dict(patches))
|
|
self.config_values.update(patches)
|
|
return {"patched": list(patches)}
|
|
|
|
async def send_message(self, session_key, message, attachments=None, elevated=None):
|
|
self.send_calls.append(
|
|
{
|
|
"session_key": session_key,
|
|
"message": message,
|
|
"attachments": attachments,
|
|
"elevated": elevated,
|
|
}
|
|
)
|
|
# Drain immediately — REPL loop sees no events and proceeds to next prompt.
|
|
if False:
|
|
yield {}
|
|
|
|
async def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
_FakeGatewayClient.instances = []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_chat_forwards_model_to_create_session(monkeypatch) -> None:
|
|
"""`opensquilla chat --model X` (gateway mode) must reach create_session(model='X')."""
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
inputs = iter(["/quit"])
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._gateway_chat(model="anthropic/claude-sonnet-4", session_id=None)
|
|
|
|
assert len(_FakeGatewayClient.instances) == 1
|
|
fake = _FakeGatewayClient.instances[-1]
|
|
assert fake.connected is True
|
|
assert fake.closed is True
|
|
assert fake.create_calls == [
|
|
{
|
|
"agent_id": "main",
|
|
"model": "anthropic/claude-sonnet-4",
|
|
"display_name": None,
|
|
}
|
|
]
|
|
assert fake.send_calls == [] # /quit on first prompt — no message sent
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_chat_session_id_skips_create_session(monkeypatch) -> None:
|
|
"""`opensquilla chat --session abc` (gateway mode) must reuse the key without create."""
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
inputs = iter(["hi", "/quit"])
|
|
|
|
async def fake_prompt_user(prefix: str = "[you] ", **kwargs):
|
|
return next(inputs)
|
|
|
|
_install_fake_inputs(monkeypatch, inputs)
|
|
|
|
await chat_cmd._gateway_chat(model=None, session_id="agent:main:resumed-key")
|
|
|
|
fake = _FakeGatewayClient.instances[-1]
|
|
assert fake.create_calls == [] # MUST NOT create
|
|
assert len(fake.send_calls) == 1
|
|
assert fake.send_calls[0]["session_key"] == "agent:main:resumed-key"
|
|
assert fake.send_calls[0]["message"] == "hi"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_slash_new_passes_title_as_display_name(monkeypatch) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
state = ChatSessionState(session_key="agent:main:old", model="openai/test")
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command(
|
|
"/new Research Notes", state, fake, {"mode": None}
|
|
)
|
|
|
|
assert handled is True
|
|
assert fake.create_calls == [
|
|
{
|
|
"agent_id": "main",
|
|
"model": "openai/test",
|
|
"display_name": "Research Notes",
|
|
}
|
|
]
|
|
assert state.session_key == "agent:main:fake12345"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_path_command_sends_prompt_without_attachments_or_upload(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
fake.is_local_gateway = True
|
|
|
|
async def fail_upload_file(*args, **kwargs):
|
|
raise AssertionError("upload_file must not be called for /path")
|
|
|
|
fake.upload_file = fail_upload_file
|
|
target = tmp_path / "large.log"
|
|
target.write_text("hello\n", encoding="utf-8")
|
|
state = ChatSessionState(session_key="agent:main:abc123", model="openai/test")
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command(
|
|
f"/path {target} summarize", state, fake, {"mode": None}
|
|
)
|
|
|
|
assert handled is True
|
|
assert len(fake.send_calls) == 1
|
|
assert fake.send_calls[0]["attachments"] == []
|
|
assert "summarize" in fake.send_calls[0]["message"]
|
|
assert str(target.resolve(strict=False)) in fake.send_calls[0]["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_path_command_remote_rejects_before_send(
|
|
monkeypatch,
|
|
tmp_path,
|
|
) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
fake.is_local_gateway = False
|
|
state = ChatSessionState(session_key="agent:main:abc123", model="openai/test")
|
|
buffer = io.StringIO()
|
|
monkeypatch.setattr(
|
|
slash_bridge,
|
|
"console",
|
|
Console(file=buffer, force_terminal=False, width=100, highlight=False),
|
|
)
|
|
nonexistent = tmp_path / "does-not-exist.log"
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command(
|
|
f"/path {nonexistent} inspect", state, fake, {"mode": None}
|
|
)
|
|
|
|
assert handled is True
|
|
assert fake.send_calls == []
|
|
assert "Use /file to upload from this CLI machine" in buffer.getvalue()
|
|
assert "File not found" not in buffer.getvalue()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_chat_does_not_forward_workspace_fields() -> None:
|
|
from opensquilla.cli.gateway_client import GatewayClient
|
|
|
|
client = GatewayClient()
|
|
calls: list[tuple[str, dict[str, object]]] = []
|
|
|
|
async def fake_call(method: str, params: dict[str, object]) -> dict[str, object]:
|
|
calls.append((method, params))
|
|
return {}
|
|
|
|
client._call = fake_call # type: ignore[method-assign]
|
|
client._recv_queue.put_nowait({"event": "session.event.done", "payload": {}})
|
|
|
|
events = [
|
|
event
|
|
async for event in client.send_message(
|
|
"agent:main:abc123",
|
|
"hello",
|
|
attachments=[],
|
|
)
|
|
]
|
|
|
|
assert events[-1]["event"] == "session.event.done"
|
|
method, params = calls[1]
|
|
assert method == "sessions.send"
|
|
source = params["_source"]
|
|
assert "workspace_dir" not in source
|
|
assert "workspace_strict" not in source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_client_resolve_approval_forwards_choice() -> None:
|
|
from opensquilla.cli.gateway_client import GatewayClient
|
|
|
|
client = GatewayClient()
|
|
calls: list[tuple[str, dict[str, object]]] = []
|
|
|
|
async def fake_call(method: str, params: dict[str, object]) -> dict[str, object]:
|
|
calls.append((method, params))
|
|
return {"status": "resolved"}
|
|
|
|
client._call = fake_call # type: ignore[method-assign]
|
|
|
|
result = await client.resolve_approval(
|
|
"approval-1",
|
|
True,
|
|
choice="allow_same_type",
|
|
)
|
|
|
|
assert result == {"status": "resolved"}
|
|
assert calls == [
|
|
(
|
|
"exec.approval.resolve",
|
|
{
|
|
"id": "approval-1",
|
|
"approved": True,
|
|
"choice": "allow_same_type",
|
|
},
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_client_follows_background_task_group_until_terminal() -> None:
|
|
from opensquilla.cli.gateway_client import GatewayClient
|
|
|
|
client = GatewayClient()
|
|
calls: list[tuple[str, dict[str, object]]] = []
|
|
|
|
async def fake_call(method: str, params: dict[str, object]) -> dict[str, object]:
|
|
calls.append((method, params))
|
|
return {}
|
|
|
|
client._call = fake_call # type: ignore[method-assign]
|
|
group_id = "subagent:agent:main:abc123:task-parent"
|
|
for frame in (
|
|
{"event": "session.event.task_group.waiting", "payload": {"group_id": group_id}},
|
|
{"event": "session.event.done", "payload": {"reason": "parent_yielded"}},
|
|
{
|
|
"event": "session.event.task_group.synthesizing",
|
|
"payload": {"group_id": group_id, "synthesis_task_id": "task-synth"},
|
|
},
|
|
{"event": "session.event.done", "payload": {"reason": "synthesis_done"}},
|
|
{"event": "task.succeeded", "payload": {"task_id": "task-synth"}},
|
|
{
|
|
"event": "session.event.task_group.done",
|
|
"payload": {"group_id": group_id, "delivery_status": "not_applicable"},
|
|
},
|
|
):
|
|
client._recv_queue.put_nowait(frame)
|
|
|
|
events = [
|
|
event
|
|
async for event in client.send_message(
|
|
"agent:main:abc123",
|
|
"hello",
|
|
attachments=[],
|
|
)
|
|
]
|
|
|
|
assert [event["event"] for event in events] == [
|
|
"session.event.task_group.waiting",
|
|
"session.event.done",
|
|
"session.event.task_group.synthesizing",
|
|
"session.event.done",
|
|
"task.succeeded",
|
|
"session.event.task_group.done",
|
|
]
|
|
assert events[-1]["delivery_status"] == "not_applicable"
|
|
assert calls[0][0] == "sessions.messages.subscribe"
|
|
assert calls[1][0] == "sessions.send"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_client_does_not_wait_for_late_task_group_after_done() -> None:
|
|
from opensquilla.cli.gateway_client import GatewayClient
|
|
|
|
client = GatewayClient()
|
|
|
|
async def fake_call(method: str, params: dict[str, object]) -> dict[str, object]:
|
|
return {}
|
|
|
|
client._call = fake_call # type: ignore[method-assign]
|
|
client._recv_queue.put_nowait({"event": "session.event.done", "payload": {}})
|
|
client._recv_queue.put_nowait(
|
|
{
|
|
"event": "session.event.task_group.synthesizing",
|
|
"payload": {"group_id": "late-group"},
|
|
}
|
|
)
|
|
|
|
events = [
|
|
event
|
|
async for event in client.send_message(
|
|
"agent:main:abc123",
|
|
"hello",
|
|
attachments=[],
|
|
)
|
|
]
|
|
|
|
assert [event["event"] for event in events] == ["session.event.done"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_client_does_not_end_on_untracked_task_group_terminal() -> None:
|
|
from opensquilla.cli.gateway_client import GatewayClient
|
|
|
|
client = GatewayClient()
|
|
|
|
async def fake_call(method: str, params: dict[str, object]) -> dict[str, object]:
|
|
return {}
|
|
|
|
client._call = fake_call # type: ignore[method-assign]
|
|
client._recv_queue.put_nowait(
|
|
{
|
|
"event": "session.event.task_group.done",
|
|
"payload": {"group_id": "untracked-group", "delivery_status": "not_applicable"},
|
|
}
|
|
)
|
|
client._recv_queue.put_nowait({"event": "session.event.done", "payload": {}})
|
|
|
|
events = [
|
|
event
|
|
async for event in client.send_message(
|
|
"agent:main:abc123",
|
|
"hello",
|
|
attachments=[],
|
|
)
|
|
]
|
|
|
|
assert [event["event"] for event in events] == [
|
|
"session.event.task_group.done",
|
|
"session.event.done",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_slash_clear_resets_session_state(monkeypatch) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
state = ChatSessionState(session_key="agent:main:abc123", model="openai/test")
|
|
state.transcript.add("user", "hello")
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command("/clear", state, fake, {"mode": None})
|
|
|
|
assert handled is True
|
|
assert fake.reset_calls == ["agent:main:abc123"]
|
|
assert state.transcript.turns == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_slash_compact_calls_session_rpc(monkeypatch) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
state = ChatSessionState(session_key="agent:main:abc123", model="openai/test")
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command("/compact", state, fake, {"mode": None})
|
|
|
|
assert handled is True
|
|
assert fake.compact_calls == [{"session_key": "agent:main:abc123"}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_slash_compact_skipped_uses_context_budget_wording(monkeypatch) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
|
|
async def compact_skipped(session_key: str) -> dict[str, object]:
|
|
fake.compact_calls.append({"session_key": session_key})
|
|
return {"key": session_key, "compacted": False}
|
|
|
|
fake.compact_session = compact_skipped
|
|
state = ChatSessionState(session_key="agent:main:abc123", model="openai/test")
|
|
buffer = io.StringIO()
|
|
monkeypatch.setattr(
|
|
slash_bridge,
|
|
"console",
|
|
Console(file=buffer, force_terminal=False, width=100, highlight=False),
|
|
)
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command("/compact", state, fake, {"mode": None})
|
|
|
|
assert handled is True
|
|
assert fake.compact_calls == [{"session_key": "agent:main:abc123"}]
|
|
output = buffer.getvalue()
|
|
assert "compact skipped" in output
|
|
assert "already within context budget; no compact was applied" in output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_slash_compact_reports_started_and_failure(monkeypatch) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
|
|
async def compact_failed(session_key: str) -> dict[str, object]:
|
|
fake.compact_calls.append({"session_key": session_key})
|
|
raise RuntimeError("provider down")
|
|
|
|
fake.compact_session = compact_failed
|
|
state = ChatSessionState(session_key="agent:main:abc123", model="openai/test")
|
|
buffer = io.StringIO()
|
|
monkeypatch.setattr(
|
|
slash_bridge,
|
|
"console",
|
|
Console(file=buffer, force_terminal=False, width=100, highlight=False),
|
|
)
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command("/compact", state, fake, {"mode": None})
|
|
|
|
assert handled is True
|
|
assert fake.compact_calls == [{"session_key": "agent:main:abc123"}]
|
|
output = buffer.getvalue()
|
|
assert "compacting context" in output
|
|
assert "compact failed: provider down" in output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_slash_delete_resolves_and_reports_errors(monkeypatch) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
fake.resolved_payload = {"session_key": "agent:main:abc123"}
|
|
fake.delete_result = {"deleted": [], "errors": ["agent:main:abc123: locked"]}
|
|
state = ChatSessionState(session_key="agent:main:current", model="openai/test")
|
|
buffer = io.StringIO()
|
|
monkeypatch.setattr(
|
|
slash_bridge,
|
|
"console",
|
|
Console(file=buffer, force_terminal=False, width=100, highlight=False),
|
|
)
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command(
|
|
"/delete abc", state, fake, {"mode": None}
|
|
)
|
|
|
|
assert handled is True
|
|
assert fake.resolve_calls == ["abc"]
|
|
assert fake.delete_calls == [["agent:main:abc123"]]
|
|
output = buffer.getvalue()
|
|
assert "Delete failed" in output
|
|
assert "locked" in output
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_slash_save_exports_persisted_history(monkeypatch, tmp_path) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
state = ChatSessionState(session_key="agent:main:abc123", model="openai/test")
|
|
output = tmp_path / "saved.md"
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command(
|
|
f"/save {output}", state, fake, {"mode": None}
|
|
)
|
|
|
|
assert handled is True
|
|
assert fake.history_calls == [{"session_key": "agent:main:abc123", "limit": 1000}]
|
|
text = output.read_text(encoding="utf-8")
|
|
assert "## You" in text
|
|
assert "persisted hello" in text
|
|
assert "## Assistant" in text
|
|
assert "persisted reply" in text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_slash_models_does_not_hit_model_prefix(monkeypatch) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
state = ChatSessionState(session_key="agent:main:abc123", model="openai/test")
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command("/models", state, fake, {"mode": None})
|
|
|
|
assert handled is True
|
|
assert fake.list_models_calls == 1
|
|
assert state.model == "openai/test"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_slash_usage_calls_usage_status(monkeypatch) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
state = ChatSessionState(session_key="agent:main:abc123", model="openai/test")
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command("/usage", state, fake, {"mode": None})
|
|
|
|
assert handled is True
|
|
assert fake.usage_status_calls == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_slash_unknown_prefix_is_not_handled(monkeypatch) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
state = ChatSessionState(session_key="agent:main:abc123", model="openai/test")
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command("/newer", state, fake, {"mode": None})
|
|
|
|
assert handled is False
|
|
assert fake.create_calls == []
|
|
assert state.session_key == "agent:main:abc123"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_stream_keyboard_interrupt_aborts_turn(monkeypatch) -> None:
|
|
class InterruptingGatewayClient(_FakeGatewayClient):
|
|
async def send_message(self, session_key, message, attachments=None, elevated=None):
|
|
self.send_calls.append(
|
|
{
|
|
"session_key": session_key,
|
|
"message": message,
|
|
"attachments": attachments,
|
|
"elevated": elevated,
|
|
}
|
|
)
|
|
raise KeyboardInterrupt
|
|
yield {}
|
|
|
|
InterruptingGatewayClient.instances = []
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", InterruptingGatewayClient)
|
|
fake = InterruptingGatewayClient()
|
|
|
|
result = await chat_cmd._stream_response_gateway(
|
|
fake,
|
|
"agent:main:abc123",
|
|
"hello",
|
|
{"mode": None},
|
|
)
|
|
|
|
assert result.cancelled is True
|
|
assert fake.abort_calls == ["agent:main:abc123"]
|
|
assert fake.send_calls[0]["message"] == "hello"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_stream_cancelled_error_aborts_turn(monkeypatch) -> None:
|
|
class CancelledGatewayClient(_FakeGatewayClient):
|
|
async def send_message(self, session_key, message, attachments=None, elevated=None):
|
|
self.send_calls.append(
|
|
{
|
|
"session_key": session_key,
|
|
"message": message,
|
|
"attachments": attachments,
|
|
"elevated": elevated,
|
|
}
|
|
)
|
|
raise asyncio.CancelledError
|
|
yield {}
|
|
|
|
CancelledGatewayClient.instances = []
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", CancelledGatewayClient)
|
|
fake = CancelledGatewayClient()
|
|
|
|
result = await chat_cmd._stream_response_gateway(
|
|
fake,
|
|
"agent:main:abc123",
|
|
"hello",
|
|
{"mode": None},
|
|
)
|
|
|
|
assert result.cancelled is True
|
|
assert fake.abort_calls == ["agent:main:abc123"]
|
|
assert fake.send_calls[0]["message"] == "hello"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_stream_renders_task_group_status_without_buffer_pollution(
|
|
monkeypatch,
|
|
) -> None:
|
|
class StatusGatewayClient(_FakeGatewayClient):
|
|
async def send_message(self, session_key, message, attachments=None, elevated=None):
|
|
yield {
|
|
"event": "session.event.task_group.waiting",
|
|
"group_id": "group-1",
|
|
"pending_count": 2,
|
|
}
|
|
yield {
|
|
"event": "session.event.task_group.synthesizing",
|
|
"group_id": "group-1",
|
|
"child_count": 2,
|
|
}
|
|
yield {"event": "session.event.text_delta", "text": "answer"}
|
|
yield {
|
|
"event": "session.event.task_group.done",
|
|
"group_id": "group-1",
|
|
"delivery_status": "not_applicable",
|
|
}
|
|
yield {"event": "session.event.done"}
|
|
|
|
class RecordingRenderer:
|
|
instances: list[RecordingRenderer] = []
|
|
|
|
def __init__(self, *_args, **_kwargs) -> None:
|
|
self.buffer = ""
|
|
self.statuses: list[str] = []
|
|
self.finalized = False
|
|
RecordingRenderer.instances.append(self)
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> bool:
|
|
return False
|
|
|
|
def append_text(self, delta: str) -> None:
|
|
self.buffer += delta
|
|
|
|
async def aappend_text(self, delta: str) -> None:
|
|
# Async sibling used by production stream paths.
|
|
self.buffer += delta
|
|
|
|
def status(self, message: str, **_kwargs) -> None:
|
|
self.statuses.append(message)
|
|
|
|
def tool_call(self, *_args, **_kwargs) -> None:
|
|
return None
|
|
|
|
def error(self, message: str) -> None:
|
|
raise AssertionError(f"unexpected error render: {message}")
|
|
|
|
def finalize(self, *_args, **_kwargs) -> None:
|
|
self.finalized = True
|
|
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.opentui.renderer.OpenTuiStreamRenderer",
|
|
RecordingRenderer,
|
|
)
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.adapters.turn_stream_defaults.OpenTuiStreamRenderer",
|
|
RecordingRenderer,
|
|
)
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.adapters.runtime_bridge.validate_tui_backend_selection",
|
|
lambda env=None: "opentui",
|
|
)
|
|
fake = StatusGatewayClient()
|
|
|
|
result = await chat_cmd._stream_response_gateway(
|
|
fake,
|
|
"agent:main:abc123",
|
|
"hello",
|
|
{"mode": None},
|
|
)
|
|
|
|
renderer = RecordingRenderer.instances[-1]
|
|
assert result.text == "answer"
|
|
assert renderer.buffer == "answer"
|
|
assert renderer.finalized is True
|
|
assert len(renderer.statuses) == 3
|
|
assert "waiting" in renderer.statuses[0]
|
|
assert "synthesizing" in renderer.statuses[1]
|
|
assert "complete" in renderer.statuses[2]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_stream_collects_artifact_events(monkeypatch) -> None:
|
|
artifact = {
|
|
"id": "art-chat",
|
|
"kind": "artifact_ref",
|
|
"name": "report.txt",
|
|
"mime": "text/plain",
|
|
"size": 4,
|
|
"sha256": "e" * 64,
|
|
"session_id": "session-1",
|
|
"session_key": "agent:main:abc123",
|
|
"source": "publish_artifact",
|
|
"created_at": "2026-05-06T12:00:00Z",
|
|
"download_url": "/api/v1/artifacts/art-chat?sessionKey=agent%3Amain%3Aabc123",
|
|
}
|
|
|
|
class ArtifactGatewayClient(_FakeGatewayClient):
|
|
async def send_message(self, session_key, message, attachments=None, elevated=None):
|
|
yield {"event": "session.event.artifact", **artifact}
|
|
yield {"event": "session.event.text_delta", "text": "answer"}
|
|
yield {"event": "session.event.done"}
|
|
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.opentui.renderer.OpenTuiStreamRenderer",
|
|
_RecordingRenderer,
|
|
)
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.adapters.turn_stream_defaults.OpenTuiStreamRenderer",
|
|
_RecordingRenderer,
|
|
)
|
|
# Renderer choice is now backend-aware; pin the OpenTUI backend so the
|
|
# recording renderer (patched above) is the one the bridge selects.
|
|
monkeypatch.setattr(
|
|
"opensquilla.cli.tui.adapters.runtime_bridge.validate_tui_backend_selection",
|
|
lambda env=None: "opentui",
|
|
)
|
|
result = await chat_cmd._stream_response_gateway(
|
|
ArtifactGatewayClient(),
|
|
"agent:main:abc123",
|
|
"hello",
|
|
{"mode": None},
|
|
)
|
|
|
|
assert result.text == "answer"
|
|
assert result.artifacts[0]["download_url"] == "/api/v1/artifacts/art-chat"
|
|
assert "session_key" not in result.artifacts[0]
|
|
assert "sessionKey" not in json.dumps(result.artifacts[0])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_elevated_unknown_prefix_is_not_handled(monkeypatch) -> None:
|
|
_FakeGatewayClient.instances.clear()
|
|
monkeypatch.setattr("opensquilla.cli.gateway_client.GatewayClient", _FakeGatewayClient)
|
|
fake = _FakeGatewayClient()
|
|
state = ChatSessionState(session_key="agent:main:abc123", model="openai/test")
|
|
|
|
handled = await chat_cmd._handle_gateway_slash_command(
|
|
"/elevatedx", state, fake, {"mode": None}
|
|
)
|
|
|
|
assert handled is False
|