Harness 是什么

class LoopCoordinator:
    """Coordinates the outer task-loop lifecycle.

    Attributes:
        current_iteration: Number of completed rounds
            (read-only).
        is_aborted: Whether abort was requested
            (read-only).
        stop_reason: Name of the evaluator that triggered
            the stop, or None if still running (read-only).
    """

    def __init__(
        self,
        evaluators: Optional[
            List[StopConditionEvaluator]
        ] = None,
    ) -> None:
        self._evaluators: List[StopConditionEvaluator] = (
            evaluators or []
        )
        self._iteration: int = 0
        self._token_usage: int = 0
        self._aborted: bool = False
        self._start_time: float = 0.0
        self._stop_reason: Optional[str] = None
        self._last_result: Optional[Dict[str, Any]] = None

def should_continue(self) -> bool:
    """Return True if the loop may proceed.

    Evaluates all evaluators with OR semantics — the first
    evaluator that returns True from ``should_stop()``
    terminates the loop and records the stop reason.
    """
    if self._aborted:
        self._stop_reason = "Aborted"
        return False

    ctx = self._build_eval_context()
    for ev in self._evaluators:
        try:
            if ev.should_stop(ctx):
                self._stop_reason = ev.name
                logger.info(
                    "Stop condition met: %s",
                    ev.name,
                )
                return False
        except Exception:
            logger.warning(
                "Evaluator %s raised an error",
                ev.name,
                exc_info=True,
            )
    return True

def get_state(self) -> Dict[str, Any]:
    """Export a JSON-safe snapshot for checkpointing."""
    ev_states: Dict[str, Any] = {}
    for ev in self._evaluators:
        s = ev.get_state()
        if s is not None:
            ev_states[ev.name] = s
    return {
        "iteration": self._iteration,
        "token_usage": self._token_usage,
        "stop_reason": self._stop_reason,
        "evaluator_states": ev_states,
    }

def load_state(
    self,
    data: Optional[Dict[str, Any]],
) -> None:
    """Restore state from a persisted snapshot.

    ``start_time`` is reset to now so that
    ``TimeoutEvaluator`` measures from the resume point.
    """
    if not data:
        return
    self._iteration = int(
        data.get("iteration", 0) or 0
    )
    self._token_usage = int(
        data.get("token_usage", 0) or 0
    )
    self._stop_reason = data.get("stop_reason")
    self._start_time = time.monotonic()
    ev_states: Dict[str, Any] = data.get(
        "evaluator_states", {}
    )
    for ev in self._evaluators:
        if ev.name in ev_states:
            ev.load_state(ev_states[ev.name])