Harness 工程深入

async def _stream_session_pipeline(
    self,
    tasks: Optional[List[OptimizationTask]] = None,
) -> AsyncIterator[Any]:
    """Run the session pipeline, yielding OutputSchema chunks."""
    started_at = time.monotonic()
    self._results = []
    self._last_cycle_result = CycleResult()
    self.artifacts = ArtifactStore()
    self._cancelled = False  # Reset cancellation state for new session
    self.budget.start()
    yield self._msg("会话启动")
    logger.info(
        "[AutoHarnessOrchestrator] session started: local_repo=%s "
        "repo_url=%s pipeline_preference=%s task_timeout=%.1fs "
        "model_timeout=%.1fs session_budget=%.1fs tasks=%s",
        ...
    )

    selected_pipeline = self._select_session_pipeline(
        tasks
    )
    self.runtime.selected_pipeline = (
        selected_pipeline.pipeline_name
    )
    logger.info(
        "[AutoHarnessOrchestrator] pipeline selected: pipeline=%s reason=%s confidence=%s",
        selected_pipeline.pipeline_name,
        getattr(selected_pipeline, "reason", ""),
        getattr(selected_pipeline, "confidence", ""),
    )
    self.artifacts.put(
        "pipeline_selection",
        selected_pipeline,
    )

logger.info(
    "[AutoHarnessOrchestrator] session finished: results=%d elapsed=%.1fs budget_remaining=%.1fs",
    len(self._results),
    time.monotonic() - started_at,
    self.budget.remaining_secs,
)
yield OutputSchema(
    type="harness_session_finished",
    index=0,
    payload={
        "pipeline": pipeline_name,
        "status": "success",
        "results_count": len(self._results),
        "is_terminal": True,
    },
)