Search for a command to run...
"""Reusable Harbor adapter for scripted or LLM-user conversations."""
from __future__ import annotations
import asyncio
import json
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from typing import Mapping, Optional, Protocol, Tuple
from harbor.agents.base import BaseAgent
from harbor.environments.base import BaseEnvironment
from harbor.models.agent.context import AgentContext
from harbor.models.trajectories import Agent, FinalMetrics, Step, Trajectory
from .model_user import ModelUser
from .runner import (
ConversationResult,
ConversationRunError,
run_llm_user_conversation,
run_scripted_conversation,
)
MAX_EVIDENCE_CHARS = 1_000_000
def _timestamp() -> str:
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
@dataclass(frozen=True)
class HarnessReply:
"""One visible response and directly observed evidence for a Harness turn."""
message: str
evidence: Mapping[str, object]
class HarnessSession(Protocol):
session_id: str
public_config: Mapping[str, object]
async def send(self, user_message: str) -> HarnessReply: ...
class RecordingSession:
"""Adapt a repository Harness session to the conversation runner."""
def __init__(self, harness: HarnessSession) -> None:
self.harness = harness
self.events: list[dict[str, object]] = []
self.exchanges: list[dict[str, object]] = []
async def send(self, user_message: str) -> str:
self.events.append(
{"role": "user", "content": user_message, "timestamp": _timestamp()}
)
try:
reply = await self.harness.send(user_message)
if not isinstance(reply, HarnessReply):
raise TypeError("Harness session must return HarnessReply")
except Exception as error:
self.exchanges.append(
{"user_message": user_message, "error": type(error).__name__}
)
raise
encoded_evidence = json.dumps(dict(reply.evidence))
if len(encoded_evidence) > MAX_EVIDENCE_CHARS:
raise ValueError("Harness evidence is too large")
evidence = json.loads(encoded_evidence)
self.events.append(
{
"role": "assistant",
"content": reply.message,
"evidence": evidence,
"timestamp": _timestamp(),
}
)
self.exchanges.append(
{
"user_message": user_message,
"assistant_message": reply.message,
"evidence": evidence,
}
)
return reply.message
class MultiTurnHarborAgent(BaseAgent):
"""Subclass this adapter and implement the repository-specific bindings."""
SUPPORTS_ATIF = True
MAX_TURNS = 8
SIMULATOR_MODEL: Optional[str] = None
@staticmethod
def name() -> str:
return "multi-turn-harness"
def version(self) -> str:
return "1.0.0"
async def setup(self, environment: BaseEnvironment) -> None:
return None
async def create_harness_session(
self, environment: BaseEnvironment, context: AgentContext
) -> HarnessSession:
raise NotImplementedError
def scripted_followups(self) -> Optional[Tuple[str, ...]]:
return None
def user_contract(self) -> Optional[str]:
return None
async def call_user_model(self, system: str, payload: str) -> str:
raise NotImplementedError
async def read_user_observation(
self, environment: BaseEnvironment
) -> Mapping[str, object]:
return {}
async def run(
self, instruction: str, environment: BaseEnvironment, context: AgentContext
) -> None:
result: Optional[ConversationResult] = None
model_user: Optional[ModelUser] = None
session: Optional[RecordingSession] = None
error_type: Optional[str] = None
contract: Optional[str] = None
run_error: Optional[BaseException] = None
try:
session = RecordingSession(
await self.create_harness_session(environment, context)
)
followups = self.scripted_followups()
contract = self.user_contract()
if (followups is None) == (contract is None):
raise ValueError("choose exactly one of scripted_followups or user_contract")
if followups is not None:
result = await run_scripted_conversation(
first_message=instruction,
followups=followups,
session=session,
)
else:
if not self.SIMULATOR_MODEL:
raise ValueError("set SIMULATOR_MODEL for an LLM user")
model_user = ModelUser(
contract=contract or "",
call_model=self.call_user_model,
read_observation=lambda: self.read_user_observation(environment),
)
result = await run_llm_user_conversation(
first_message=instruction,
session=session,
simulated_user=model_user,
max_turns=self.MAX_TURNS,
)
except ConversationRunError as error:
result = error.result
error_type = type(error).__name__
run_error = error
except asyncio.CancelledError as error:
error_type = type(error).__name__
run_error = error
except Exception as error:
error_type = type(error).__name__
run_error = error
interaction = self._interaction(
instruction, session, result, model_user, contract, error_type
)
artifact_errors: list[dict[str, str]] = []
interaction_path = self.logs_dir / "interaction.json"
trajectory_path = self.logs_dir / "trajectory.json"
try:
self.logs_dir.mkdir(parents=True, exist_ok=True)
interaction_path.write_text(
json.dumps(interaction, indent=2), encoding="utf-8"
)
except Exception as error:
artifact_errors.append(
{"operation": "write interaction", "error": type(error).__name__}
)
try:
trajectory_path.write_text(
json.dumps(self._trajectory(interaction).to_json_dict(), indent=2),
encoding="utf-8",
)
except Exception as error:
artifact_errors.append(
{"operation": "write trajectory", "error": type(error).__name__}
)
if artifact_errors and interaction_path.is_file():
interaction["artifact_errors"] = artifact_errors
try:
interaction_path.write_text(
json.dumps(interaction, indent=2), encoding="utf-8"
)
except Exception:
pass
if run_error is not None:
raise run_error.with_traceback(run_error.__traceback__)
if artifact_errors:
operations = ", ".join(item["operation"] for item in artifact_errors)
raise RuntimeError(f"artifact handling failed: {operations}")
def _interaction(
self,
instruction: str,
session: Optional[RecordingSession],
result: Optional[ConversationResult],
model_user: Optional[ModelUser],
contract: Optional[str],
error_type: Optional[str],
) -> dict[str, object]:
harness = session.harness if session else None
turns = [asdict(turn) for turn in result.turns] if result else []
if turns and session:
exchanges = iter(session.exchanges)
for turn in turns:
if turn["role"] != "assistant":
continue
exchange = next(exchanges, {})
if "evidence" in exchange:
turn["evidence"] = exchange["evidence"]
return {
"instruction": instruction,
"session_id": harness.session_id if harness else None,
"harness_config": json.loads(json.dumps(dict(harness.public_config)))
if harness
else None,
"turns": turns or (session.events if session else []),
"harness_exchanges": session.exchanges if session else [],
"simulation": {
"mode": "llm_user" if contract is not None else "scripted",
"model": self.SIMULATOR_MODEL if contract is not None else None,
"user_contract": contract,
},
"simulator_decisions": model_user.records if model_user else [],
"termination": result.termination if result else "error",
"error": {"type": error_type} if error_type else None,
}
def _trajectory(self, interaction: Mapping[str, object]) -> Trajectory:
raw_turns = interaction["turns"]
turns = raw_turns if isinstance(raw_turns, list) else []
steps: list[Step] = []
for index, turn in enumerate(turns, 1):
role = turn["role"]
extra = {
key: value
for key, value in turn.items()
if key not in {"role", "content", "timestamp"} and value is not None
}
values = {
"step_id": index,
"source": "agent" if role == "assistant" else "user",
"message": turn["content"],
"extra": extra or None,
}
if turn.get("timestamp"):
values["timestamp"] = turn["timestamp"]
steps.append(Step(**values))
if not steps:
steps.append(Step(step_id=1, source="user", message=interaction["instruction"]))
return Trajectory(
schema_version="ATIF-v1.7",
session_id=interaction.get("session_id"),
agent=Agent(name=self.name(), version=self.version()),
steps=steps,
final_metrics=FinalMetrics(total_steps=len(steps)),
extra={"termination": interaction["termination"], "error": interaction["error"]},
)