File size: 5,491 Bytes
0b2c941 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 | # SPDX-License-Identifier: BSD-3-Clause
"""Turn-based adapter interface for external agentic harnesses (RFC 005)."""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import AsyncIterator, Optional
from ..env_server.mcp_types import Tool
from .config import HarnessConfig
from .events import HarnessEvent, HarnessEventType, HarnessResponse
class HarnessError(Exception):
"""Base error for agentic harness failures."""
class HarnessStartupError(HarnessError):
"""The harness process failed to start or become ready in time."""
class HarnessNotRunningError(HarnessError):
"""An operation required a running harness but none was alive."""
class HarnessTurnTimeoutError(HarnessError):
"""A single conversational turn exceeded its wall-clock budget."""
class AgenticHarnessAdapter(ABC):
"""
Abstract adapter for a turn-based external agentic harness.
Subclass this to integrate a harness such as OpenClaw or Claude Code. The
adapter owns the harness process lifecycle and communication; the harness
itself is a long-lived process that maintains conversation context across
turns. Each `send_message()` call is one conversational turn: the harness
runs its internal ReAct loop and returns when it has a response.
Distinct from the trainer-side rollout
[`~openenv.core.harness.rollout.HarnessAdapter`], which drives an entire
episode in a single call.
Attributes:
BUILTIN_TOOL_NAMES (`frozenset[str]`):
Names of the harness's built-in tools, used to detect conflicts
with injected environment tools. Concrete adapters override this.
"""
BUILTIN_TOOL_NAMES: frozenset[str] = frozenset()
def __init__(self, config: HarnessConfig):
self.config = config
@abstractmethod
async def start(self, working_directory: str) -> None:
"""
Start the harness process.
Args:
working_directory (`str`):
Path where the harness should operate.
Raises:
[`~openenv.core.harness.adapter.HarnessStartupError`]:
If the process fails to start or become ready in time.
"""
...
@abstractmethod
async def stop(self) -> None:
"""
Stop the harness process and clean up resources.
Implementations must be idempotent: calling `stop()` on an already
stopped (or never started) adapter must succeed silently.
"""
...
@abstractmethod
async def inject_tools(
self, tools: list[Tool], bridge_url: Optional[str] = None
) -> None:
"""
Inject environment MCP tool definitions into the harness configuration.
Called before `start()` so the harness discovers the tools at startup.
The mechanism is harness-specific (config file, CLI flags, environment
variables).
Args:
tools (`list[Tool]`):
Conflict-resolved environment tool definitions to inject.
bridge_url (`str`, *optional*):
URL of the MCP bridge serving these tools, when the
environment exposes one.
"""
...
@abstractmethod
def send_message_streaming(self, message: str) -> AsyncIterator[HarnessEvent]:
"""
Send a message and stream intermediate events for one turn.
Yields events as the harness processes the turn (tool calls, LLM
chunks, text output). The final event must be a
[`~openenv.core.harness.events.HarnessEvent`] of type `TURN_COMPLETE`
whose data carries the turn's `response` and optionally `done`.
Args:
message (`str`):
The user message for this conversational turn.
Yields:
[`~openenv.core.harness.events.HarnessEvent`] instances.
"""
...
@abstractmethod
async def is_alive(self) -> bool:
"""Check whether the harness process is still running."""
...
async def send_message(self, message: str) -> HarnessResponse:
"""
Send a message and collect the complete response for one turn.
Drains `send_message_streaming()` and assembles a
[`~openenv.core.harness.events.HarnessResponse`] from the terminal
`TURN_COMPLETE` event.
Args:
message (`str`):
The user message for this conversational turn.
Returns:
[`~openenv.core.harness.events.HarnessResponse`] with the text
response, all turn events, and the harness's done signal.
Raises:
[`~openenv.core.harness.adapter.HarnessError`]:
If the event stream ends without a `TURN_COMPLETE` event.
"""
events: list[HarnessEvent] = []
async for event in self.send_message_streaming(message):
events.append(event)
if not events or events[-1].type is not HarnessEventType.TURN_COMPLETE:
raise HarnessError(
"harness event stream ended without a TURN_COMPLETE event"
)
terminal = events[-1]
return HarnessResponse(
response=str(terminal.data.get("response", "")),
events=events,
done=bool(terminal.data.get("done", False)),
)
__all__ = [
"AgenticHarnessAdapter",
"HarnessError",
"HarnessNotRunningError",
"HarnessStartupError",
"HarnessTurnTimeoutError",
]
|