wehub-resource-sync
449 lines
14 KiB
Python
449 lines
14 KiB
Python
"""Runtime adapter behavior: native terminal output/interrupts, standalone
|
|
dispatch model bookkeeping and error recovery, and legacy compat renderer
|
|
selection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
import signal
|
|
import sys
|
|
from types import SimpleNamespace
|
|
from typing import Any, cast
|
|
|
|
import pytest
|
|
from rich.console import Console
|
|
|
|
from opensquilla.cli.chat.turn import TurnResult, UsageSummary
|
|
from opensquilla.cli.tui.adapters import native_bridge, slash_standalone
|
|
from opensquilla.cli.tui.backend.contracts import TuiOutputHandle
|
|
from opensquilla.engine.commands import Surface
|
|
|
|
|
|
class _FakeSessionManager:
|
|
async def get_or_create(self, session_key: str, agent_id: str = "main") -> object:
|
|
return SimpleNamespace(session_key=session_key, agent_id=agent_id)
|
|
|
|
|
|
class _FakeServices:
|
|
def __init__(self) -> None:
|
|
self.config = None
|
|
self.session_manager = _FakeSessionManager()
|
|
|
|
async def close(self) -> None:
|
|
return None
|
|
|
|
|
|
class _RecordingConsole:
|
|
def __init__(self) -> None:
|
|
self.payloads: list[Any] = []
|
|
|
|
def print(self, *args: Any, **kwargs: Any) -> None:
|
|
self.payloads.extend(args)
|
|
|
|
|
|
def _standalone_deps(
|
|
*,
|
|
stream_response: Any,
|
|
run_concurrent_repl: Any,
|
|
output_console: Any,
|
|
error_panel_factory: Any,
|
|
) -> Any:
|
|
from opensquilla.cli.tui import standalone_runtime
|
|
|
|
return standalone_runtime.StandaloneRuntimeDependencies(
|
|
stream_response=stream_response,
|
|
image_command_handler=stream_response,
|
|
run_concurrent_repl=run_concurrent_repl,
|
|
slash_services_factory=lambda _svc: slash_standalone.StandaloneSlashServices(),
|
|
sync_slash_adapter_io=lambda: None,
|
|
get_tui_output=lambda _scope: None,
|
|
output_console=output_console,
|
|
error_panel_factory=error_panel_factory,
|
|
)
|
|
|
|
|
|
def _patch_standalone_services(monkeypatch: pytest.MonkeyPatch) -> _FakeServices:
|
|
services = _FakeServices()
|
|
|
|
async def fake_build_services() -> _FakeServices:
|
|
return services
|
|
|
|
monkeypatch.setattr("opensquilla.gateway.build_services", fake_build_services)
|
|
monkeypatch.setattr(
|
|
"opensquilla.gateway.build_turn_runner_from_services",
|
|
lambda _services: object(),
|
|
)
|
|
return services
|
|
|
|
|
|
# --- Native terminal output ---------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_native_write_through_streams_midline_chunks_without_rewrapping(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
# Streamed flushes land mid-line; Rich must not word-wrap each chunk as if
|
|
# the cursor were at column 0, or wrap points misalign and break paragraphs.
|
|
narrow = Console(width=20)
|
|
monkeypatch.setattr(native_bridge, "console", narrow)
|
|
handle = native_bridge.NativeTerminalOutputHandle(
|
|
approval_surface=Surface.CLI_STANDALONE,
|
|
)
|
|
chunks = ["Hello ", "world ", "this is a much longer sentence streamed in chunks"]
|
|
|
|
with narrow.capture() as capture:
|
|
for chunk in chunks:
|
|
await handle.write_through(chunk)
|
|
|
|
assert capture.get() == "".join(chunks)
|
|
|
|
|
|
# --- Native interrupt fallback -------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_interrupt_fallback_owns_process_sigint_and_restores(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
# Where loop signal handlers are unavailable (e.g. Windows), the surface
|
|
# must own the process-level SIGINT handler for the session so Ctrl+C
|
|
# cancels the turn instead of unwinding asyncio.run, then restore it.
|
|
surface = native_bridge.NativeTerminalSurface(approval_surface=Surface.CLI_STANDALONE)
|
|
loop = asyncio.get_running_loop()
|
|
|
|
def _unsupported(*_args: Any, **_kwargs: Any) -> None:
|
|
raise NotImplementedError
|
|
|
|
monkeypatch.setattr(loop, "add_signal_handler", _unsupported)
|
|
cancelled: list[bool] = []
|
|
shutdown: list[bool] = []
|
|
surface.set_cancel_callback(lambda: cancelled.append(True))
|
|
surface.set_shutdown_callback(lambda: shutdown.append(True))
|
|
before = signal.getsignal(signal.SIGINT)
|
|
|
|
try:
|
|
surface.install_interrupt_handler()
|
|
installed = signal.getsignal(signal.SIGINT)
|
|
assert installed is not before
|
|
assert callable(installed)
|
|
installed(signal.SIGINT, None) # simulate signal delivery
|
|
await asyncio.sleep(0.01)
|
|
assert cancelled == [True]
|
|
assert shutdown == []
|
|
finally:
|
|
surface.remove_interrupt_handler()
|
|
|
|
assert signal.getsignal(signal.SIGINT) is before
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(
|
|
sys.platform.startswith("win"),
|
|
reason="os.kill-based SIGINT delivery is POSIX-only in tests",
|
|
)
|
|
async def test_interrupt_fallback_handles_real_sigint_without_exiting(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
surface = native_bridge.NativeTerminalSurface(approval_surface=Surface.CLI_STANDALONE)
|
|
loop = asyncio.get_running_loop()
|
|
|
|
def _unsupported(*_args: Any, **_kwargs: Any) -> None:
|
|
raise NotImplementedError
|
|
|
|
monkeypatch.setattr(loop, "add_signal_handler", _unsupported)
|
|
cancelled: list[bool] = []
|
|
shutdown: list[bool] = []
|
|
surface.set_cancel_callback(lambda: cancelled.append(True))
|
|
surface.set_shutdown_callback(lambda: shutdown.append(True))
|
|
before = signal.getsignal(signal.SIGINT)
|
|
|
|
try:
|
|
surface.install_interrupt_handler()
|
|
os.kill(os.getpid(), signal.SIGINT)
|
|
await asyncio.sleep(0.05)
|
|
assert cancelled == [True]
|
|
assert shutdown == []
|
|
finally:
|
|
surface.remove_interrupt_handler()
|
|
|
|
assert signal.getsignal(signal.SIGINT) is before
|
|
|
|
|
|
# --- Standalone dispatch model bookkeeping --------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_dispatch_never_feeds_routed_model_back(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
# Without --model every turn must dispatch with model=None so per-turn
|
|
# routing stays live; the routed result is display-only state.
|
|
from opensquilla.cli.tui import standalone_runtime
|
|
|
|
_patch_standalone_services(monkeypatch)
|
|
models_seen: list[str | None] = []
|
|
captured: dict[str, Any] = {}
|
|
|
|
async def fake_stream_response(
|
|
turn_runner: object,
|
|
session_key: str,
|
|
tool_ctx: object,
|
|
message: str,
|
|
model: str | None = None,
|
|
svc: object = None,
|
|
timeout: float | None = None,
|
|
*,
|
|
tui_output: TuiOutputHandle | None = None,
|
|
pending_input_provider: object | None = None,
|
|
) -> TurnResult:
|
|
models_seen.append(model)
|
|
return TurnResult(
|
|
text="reply",
|
|
usage=UsageSummary(input_tokens=1, output_tokens=1),
|
|
model_after="router/turn-pick",
|
|
)
|
|
|
|
async def fake_run_concurrent_repl(
|
|
*,
|
|
surface: Surface,
|
|
scope: standalone_runtime.StandaloneRuntimeScope,
|
|
dispatch: Any,
|
|
) -> None:
|
|
assert await dispatch("first") is True
|
|
assert await dispatch("second") is True
|
|
captured["scope_model"] = scope["model"]
|
|
|
|
deps = _standalone_deps(
|
|
stream_response=fake_stream_response,
|
|
run_concurrent_repl=fake_run_concurrent_repl,
|
|
output_console=_RecordingConsole(),
|
|
error_panel_factory=lambda message: message,
|
|
)
|
|
|
|
await standalone_runtime.run_standalone_chat(
|
|
model=None,
|
|
session_id="agent:main:standalone:test",
|
|
deps=deps,
|
|
)
|
|
|
|
assert models_seen == [None, None]
|
|
# The routed model is still mirrored to display state for the HUD.
|
|
assert captured["scope_model"] == "router/turn-pick"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_requested_model_follows_model_command_not_routing(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.tui import standalone_runtime
|
|
|
|
_patch_standalone_services(monkeypatch)
|
|
models_seen: list[str | None] = []
|
|
|
|
async def fake_stream_response(
|
|
turn_runner: object,
|
|
session_key: str,
|
|
tool_ctx: object,
|
|
message: str,
|
|
model: str | None = None,
|
|
svc: object = None,
|
|
timeout: float | None = None,
|
|
*,
|
|
tui_output: TuiOutputHandle | None = None,
|
|
pending_input_provider: object | None = None,
|
|
) -> TurnResult:
|
|
models_seen.append(model)
|
|
return TurnResult(
|
|
text="reply",
|
|
usage=UsageSummary(input_tokens=1, output_tokens=1),
|
|
model_after="router/other",
|
|
)
|
|
|
|
async def fake_run_concurrent_repl(
|
|
*,
|
|
surface: Surface,
|
|
scope: standalone_runtime.StandaloneRuntimeScope,
|
|
dispatch: Any,
|
|
) -> None:
|
|
assert await dispatch("first") is True
|
|
assert await dispatch("second") is True
|
|
assert await dispatch("/model user/second") is True
|
|
assert await dispatch("third") is True
|
|
|
|
deps = _standalone_deps(
|
|
stream_response=fake_stream_response,
|
|
run_concurrent_repl=fake_run_concurrent_repl,
|
|
output_console=_RecordingConsole(),
|
|
error_panel_factory=lambda message: message,
|
|
)
|
|
|
|
await standalone_runtime.run_standalone_chat(
|
|
model="user/requested",
|
|
session_id="agent:main:standalone:test",
|
|
deps=deps,
|
|
)
|
|
|
|
# --model survives routed turns; only /model changes the dispatch model.
|
|
assert models_seen == ["user/requested", "user/requested", "user/second"]
|
|
|
|
|
|
# --- Standalone dispatch error recovery ------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_dispatch_maps_stream_errors_to_error_panel(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.tui import standalone_runtime
|
|
|
|
_patch_standalone_services(monkeypatch)
|
|
output_console = _RecordingConsole()
|
|
messages: list[str] = []
|
|
|
|
async def flaky_stream_response(
|
|
turn_runner: object,
|
|
session_key: str,
|
|
tool_ctx: object,
|
|
message: str,
|
|
model: str | None = None,
|
|
svc: object = None,
|
|
timeout: float | None = None,
|
|
*,
|
|
tui_output: TuiOutputHandle | None = None,
|
|
pending_input_provider: object | None = None,
|
|
) -> TurnResult:
|
|
messages.append(message)
|
|
if len(messages) == 1:
|
|
raise RuntimeError("provider exploded")
|
|
return TurnResult(
|
|
text="reply",
|
|
usage=UsageSummary(input_tokens=1, output_tokens=1),
|
|
model_after=None,
|
|
)
|
|
|
|
async def fake_run_concurrent_repl(
|
|
*,
|
|
surface: Surface,
|
|
scope: standalone_runtime.StandaloneRuntimeScope,
|
|
dispatch: Any,
|
|
) -> None:
|
|
# The failing turn is reported and the loop keeps accepting input.
|
|
assert await dispatch("boom") is True
|
|
assert await dispatch("recover") is True
|
|
|
|
deps = _standalone_deps(
|
|
stream_response=flaky_stream_response,
|
|
run_concurrent_repl=fake_run_concurrent_repl,
|
|
output_console=output_console,
|
|
error_panel_factory=lambda message: f"<panel {message}>",
|
|
)
|
|
|
|
await standalone_runtime.run_standalone_chat(
|
|
model=None,
|
|
session_id="agent:main:standalone:test",
|
|
deps=deps,
|
|
)
|
|
|
|
assert messages == ["boom", "recover"]
|
|
assert "<panel RuntimeError: provider exploded>" in output_console.payloads
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_standalone_dispatch_maps_slash_errors_to_error_panel(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.tui import standalone_runtime
|
|
|
|
_patch_standalone_services(monkeypatch)
|
|
output_console = _RecordingConsole()
|
|
|
|
async def broken_slash_handler(
|
|
cmd: str,
|
|
context: slash_standalone.StandaloneSlashContext,
|
|
) -> bool:
|
|
raise OSError("db locked")
|
|
|
|
monkeypatch.setattr(
|
|
slash_standalone,
|
|
"handle_standalone_slash_command",
|
|
broken_slash_handler,
|
|
)
|
|
|
|
async def fake_stream_response(
|
|
turn_runner: object,
|
|
session_key: str,
|
|
tool_ctx: object,
|
|
message: str,
|
|
model: str | None = None,
|
|
svc: object = None,
|
|
timeout: float | None = None,
|
|
*,
|
|
tui_output: TuiOutputHandle | None = None,
|
|
pending_input_provider: object | None = None,
|
|
) -> TurnResult:
|
|
return TurnResult(
|
|
text="reply",
|
|
usage=UsageSummary(input_tokens=1, output_tokens=1),
|
|
model_after=None,
|
|
)
|
|
|
|
async def fake_run_concurrent_repl(
|
|
*,
|
|
surface: Surface,
|
|
scope: standalone_runtime.StandaloneRuntimeScope,
|
|
dispatch: Any,
|
|
) -> None:
|
|
assert await dispatch("/new") is True
|
|
assert await dispatch("still alive") is True
|
|
|
|
deps = _standalone_deps(
|
|
stream_response=fake_stream_response,
|
|
run_concurrent_repl=fake_run_concurrent_repl,
|
|
output_console=output_console,
|
|
error_panel_factory=lambda message: f"<panel {message}>",
|
|
)
|
|
|
|
await standalone_runtime.run_standalone_chat(
|
|
model=None,
|
|
session_id="agent:main:standalone:test",
|
|
deps=deps,
|
|
)
|
|
|
|
assert "<panel OSError: db locked>" in output_console.payloads
|
|
|
|
|
|
# --- Legacy compat renderer selection --------------------------------------------
|
|
|
|
|
|
def test_chat_compat_default_deps_follow_selected_backend(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from opensquilla.cli.tui.adapters import chat_compat, runtime_bridge
|
|
from opensquilla.cli.tui.native.renderer import NativeStreamRenderer
|
|
from opensquilla.cli.tui.opentui.renderer import OpenTuiStreamRenderer
|
|
|
|
monkeypatch.setattr(
|
|
runtime_bridge,
|
|
"validate_tui_backend_selection",
|
|
lambda env=None: "native",
|
|
)
|
|
deps = chat_compat.default_turn_stream_dependencies()
|
|
assert deps.renderer_factory is NativeStreamRenderer
|
|
assert deps.stream_wrapper is cast(Any, chat_compat.wrap_cli_turn_stream)
|
|
|
|
monkeypatch.setattr(
|
|
runtime_bridge,
|
|
"validate_tui_backend_selection",
|
|
lambda env=None: "opentui",
|
|
)
|
|
deps = chat_compat.default_turn_stream_dependencies()
|
|
assert deps.renderer_factory is OpenTuiStreamRenderer
|
|
|
|
|
|
def test_chat_compat_no_longer_exports_tool_compress_wrapper() -> None:
|
|
from opensquilla.cli.tui.adapters import chat_compat
|
|
|
|
assert not hasattr(chat_compat, "handle_tool_compress_command")
|