hkuds--openharness
tjb-tech
5dd8b952ec
A lightweight open-source Python implementation of the Agent Harness architecture. 44x lighter than Claude Code (11K vs 512K lines), 98% core tool coverage. - 43 tools with Pydantic validation and parallel execution - Skills system compatible with anthropics/skills (17+ tested) - Plugin system compatible with claude-code/plugins (12+ tested) - API retry with exponential backoff - Multi-level permissions with path rules - React/Ink TUI with "Oh my Harness!" branding - 114 unit tests + 6 E2E test suites - MIT License
93 行
3.1 KiB
Python
93 行
3.1 KiB
Python
"""Tests for the React backend host protocol."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from openharness.api.client import ApiMessageCompleteEvent
|
|
from openharness.api.usage import UsageSnapshot
|
|
from openharness.engine.messages import ConversationMessage, TextBlock
|
|
from openharness.ui.backend_host import BackendHostConfig, ReactBackendHost
|
|
from openharness.ui.runtime import build_runtime, close_runtime, start_runtime
|
|
|
|
|
|
class StaticApiClient:
|
|
"""Fake streaming client for backend host tests."""
|
|
|
|
def __init__(self, text: str) -> None:
|
|
self._text = text
|
|
|
|
async def stream_message(self, request):
|
|
del request
|
|
yield ApiMessageCompleteEvent(
|
|
message=ConversationMessage(role="assistant", content=[TextBlock(text=self._text)]),
|
|
usage=UsageSnapshot(input_tokens=2, output_tokens=3),
|
|
stop_reason=None,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backend_host_processes_command(tmp_path, monkeypatch):
|
|
monkeypatch.chdir(tmp_path)
|
|
monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
|
|
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
|
|
|
|
host = ReactBackendHost(BackendHostConfig(api_client=StaticApiClient("unused")))
|
|
host._bundle = await build_runtime(api_client=StaticApiClient("unused"))
|
|
events = []
|
|
|
|
async def _emit(event):
|
|
events.append(event)
|
|
|
|
host._emit = _emit # type: ignore[method-assign]
|
|
await start_runtime(host._bundle)
|
|
try:
|
|
should_continue = await host._process_line("/version")
|
|
finally:
|
|
await close_runtime(host._bundle)
|
|
|
|
assert should_continue is True
|
|
assert any(event.type == "transcript_item" and event.item and event.item.role == "user" for event in events)
|
|
assert any(
|
|
event.type == "transcript_item"
|
|
and event.item
|
|
and event.item.role == "system"
|
|
and "OpenHarness" in event.item.text
|
|
for event in events
|
|
)
|
|
assert any(event.type == "state_snapshot" for event in events)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backend_host_processes_model_turn(tmp_path, monkeypatch):
|
|
monkeypatch.chdir(tmp_path)
|
|
monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
|
|
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
|
|
|
|
host = ReactBackendHost(BackendHostConfig(api_client=StaticApiClient("hello from react backend")))
|
|
host._bundle = await build_runtime(api_client=StaticApiClient("hello from react backend"))
|
|
events = []
|
|
|
|
async def _emit(event):
|
|
events.append(event)
|
|
|
|
host._emit = _emit # type: ignore[method-assign]
|
|
await start_runtime(host._bundle)
|
|
try:
|
|
should_continue = await host._process_line("hi")
|
|
finally:
|
|
await close_runtime(host._bundle)
|
|
|
|
assert should_continue is True
|
|
assert any(
|
|
event.type == "assistant_complete" and event.message == "hello from react backend"
|
|
for event in events
|
|
)
|
|
assert any(
|
|
event.type == "assistant_complete"
|
|
and event.item
|
|
and event.item.role == "assistant"
|
|
and "hello from react backend" in event.item.text
|
|
for event in events
|
|
)
|