1.openharness源码解析。参考的文章：

花一天读完 OpenHarness：一个 11,733 行的 Agent Harness 到底长什么样
https://www.joyehuang.me/blog/20260410---openharnessphase1/post

OpenHarness 架构说明文档
https://juejin.cn/post/7627340015571959850

Harness 将大模型从一个"纯推理引擎"转变为一个"感知-推理-行动"的智能体（agent）。它填补的是行动能力和环境反馈——让模型能够验证自己的猜想、修正自己的错误、在真实状态上迭代。
 大模型 + Harness = 推理 + 行动 = 智能体。

2.是通过在源码中加log，来获取数据流的。

通过数据流，可以看到每次交互的真实数据，可以看到大模型缓存命中的原因是因为agent每次都要发送前面的会话内容。

3.一个是在run_query中，一个是在https中的加密解密出。

/home/claw/OpenHarness/src/openharness/engine/query.py

_run_query_entry_count = 0


async def run_query(
    context: QueryContext,
    messages: list[ConversationMessage],
) -> AsyncIterator[tuple[StreamEvent, UsageSnapshot | None]]:
    """Run the conversation loop until the model stops requesting tools.

    Auto-compaction is checked at the start of each turn.  When the
    estimated token count exceeds the model's auto-compact threshold,
    the engine first tries a cheap microcompact (clearing old tool result
    content) and, if that is not enough, performs a full LLM-based
    summarization of older messages.
    """
    global _run_query_entry_count
    _run_query_entry_count += 1
    with open("/home/claw/testclaude/run_query_total.txt", "a") as _f:
        _f.write(f"\n{'#' * 80}\n")
        _f.write(f"### run_query 入口 — 第 {_run_query_entry_count} 次进入 ###\n")
        _f.write(f"{'#' * 80}\n")

    from openharness.services.compact import (
        AutoCompactState,
        auto_compact_if_needed,
    )

    compact_state = AutoCompactState()
    reactive_compact_attempted = False
    last_compaction_result: tuple[list[ConversationMessage], bool] = (messages, False)
    effective_max_tokens = _bounded_completion_tokens(
        context.max_tokens,
        context.context_window_tokens,
    )
    reported_token_clamp = False

    async def _stream_compaction(
        *,
        trigger: str,
        force: bool = False,
    ) -> AsyncIterator[tuple[StreamEvent, UsageSnapshot | None]]:
        nonlocal last_compaction_result
        progress_queue: asyncio.Queue[CompactProgressEvent] = asyncio.Queue()

        async def _progress(event: CompactProgressEvent) -> None:
            await progress_queue.put(event)

        task = asyncio.create_task(
            auto_compact_if_needed(
                messages,
                api_client=context.api_client,
                model=context.model,
                system_prompt=context.system_prompt,
                state=compact_state,
                progress_callback=_progress,
                force=force,
                trigger=trigger,
                hook_executor=context.hook_executor,
                carryover_metadata=context.tool_metadata,
                context_window_tokens=context.context_window_tokens,
                auto_compact_threshold_tokens=context.auto_compact_threshold_tokens,
            )
        )
        while True:
            try:
                event = await asyncio.wait_for(progress_queue.get(), timeout=0.05)
                yield event, None
            except asyncio.TimeoutError:
                if task.done():
                    break
                continue
        while not progress_queue.empty():
            yield progress_queue.get_nowait(), None
        last_compaction_result = await task
        return

    turn_count = 0
    while context.max_turns is None or turn_count < context.max_turns:
        turn_count += 1
        with open("/home/claw/testclaude/run_query_total.txt", "a") as _f:
            _f.write(f"\n>>> TURN {turn_count} >>>\n")
        if effective_max_tokens != context.max_tokens and not reported_token_clamp:
            reported_token_clamp = True
            yield StatusEvent(
                message=(
                    "Requested max_tokens="
                    f"{context.max_tokens} exceeds the safe per-request output cap; "
                    f"using {effective_max_tokens}."
                )
            ), None
        # --- auto-compact check before calling the model ---------------
        async for event, usage in _stream_compaction(trigger="auto"):
            yield event, usage
        compacted_messages, was_compacted = last_compaction_result
        if compacted_messages is not messages:
            messages[:] = compacted_messages
        # ---------------------------------------------------------------

        # --- image preprocessing: convert ImageBlocks to text for non-vision models ---
        async for event in _preprocess_images_in_messages(messages, context):
            yield event, None
        # -----------------------------------------------------------------------------

        final_message: ConversationMessage | None = None
        usage = UsageSnapshot()

        # --- DEBUG: log API request to file (发送数据) ---
        if True:
            import json as _json
            _tools = context.tool_registry.to_api_schema()
            _req = ApiMessageRequest(
                model=context.model,
                messages=messages,
                system_prompt=context.system_prompt,
                max_tokens=effective_max_tokens,
                tools=_tools,
                effort=context.effort,
            )
            with open("/home/claw/testclaude/run_query_total.txt", "a") as _f:
                _f.write("\n" + "=" * 80 + "\n")
                _f.write(">>> 发送数据 - API请求 <<<\n")
                _f.write(f"TURN {turn_count}\n")
                _f.write(f"MODEL: {_req.model}\n")
                _f.write(f"MAX_TOKENS: {_req.max_tokens}\n")
                _f.write(f"EFFORT: {_req.effort}\n")
                _f.write(f"SYSTEM_PROMPT ({len(_req.system_prompt) if _req.system_prompt else 0} chars):\n")
                _f.write((_req.system_prompt or "(none)") + "\n")
                _f.write(f"TOOLS ({len(_tools)}):\n")
                for _t in _tools:
                    _f.write(f"    - {_t.get('name', '?')}: {_t.get('description', '')}\n")
                _f.write(f"MESSAGES ({len(messages)}):\n")
                for _i, _m in enumerate(messages):
                    _f.write(f"    [{_i}] role={_m.role!r} content_blocks={len(getattr(_m, 'content', []))}\n")
                    for _j, _b in enumerate(getattr(_m, 'content', [])):
                        _bt = type(_b).__name__
                        if hasattr(_b, 'text'):
                            _txt = getattr(_b, 'text', '')
                            _f.write(f"        [{_j}] {_bt} text={_txt!r}\n")
                        elif hasattr(_b, 'content'):
                            _cnt = getattr(_b, 'content', '')
                            _f.write(f"        [{_j}] {_bt} content={_cnt!r}\n")
                        elif hasattr(_b, 'name') and hasattr(_b, 'input'):
                            _f.write(f"        [{_j}] {_bt} name={getattr(_b, 'name', '?')!r} input={_json.dumps(getattr(_b, 'input', {}), ensure_ascii=False)!r}\n")
                        else:
                            _f.write(f"        [{_j}] {_bt}\n")
                _f.write("=" * 80 + "\n")
        # --- end debug ---

        try:
            async for event in context.api_client.stream_message(
                ApiMessageRequest(
                    model=context.model,
                    messages=messages,
                    system_prompt=context.system_prompt,
                    max_tokens=effective_max_tokens,
                    tools=context.tool_registry.to_api_schema(),
                    effort=context.effort,
                )
            ):
                # --- DEBUG: log stream events to file (接收数据) ---
                if True:
                    import json as _json
                    with open("/home/claw/testclaude/run_query_total.txt", "a") as _f:
                        if isinstance(event, ApiTextDeltaEvent):
                            _f.write(f"<<< 接收数据 - TEXT_DELTA: {event.text!r}\n")
                        elif isinstance(event, ApiRetryEvent):
                            _f.write(f"<<< 接收数据 - RETRY: attempt={event.attempt}/{event.max_attempts} delay={event.delay_seconds:.1f}s msg={event.message!r}\n")
                        elif isinstance(event, ApiMessageCompleteEvent):
                            _f.write(f"<<< 接收数据 - MESSAGE_COMPLETE: stop_reason={event.stop_reason!r}\n")
                            _f.write(f"<<< 接收数据 - USAGE: input={event.usage.input_tokens} output={event.usage.output_tokens}\n")
                            _msg = event.message
                            _f.write(f"<<< 接收数据 - FINAL_MESSAGE role={_msg.role!r} content_blocks={len(getattr(_msg, 'content', []))}\n")
                            for _j, _b in enumerate(getattr(_msg, 'content', [])):
                                _bt = type(_b).__name__
                                if hasattr(_b, 'text'):
                                    _txt = getattr(_b, 'text', '')
                                    _f.write(f"    [{_j}] {_bt} text={_txt!r}\n")
                                elif hasattr(_b, 'name') and hasattr(_b, 'input'):
                                    _f.write(f"    [{_j}] {_bt} name={getattr(_b, 'name', '?')!r} input={_json.dumps(getattr(_b, 'input', {}), ensure_ascii=False)!r}\n")
                                elif hasattr(_b, 'name'):
                                    _f.write(f"    [{_j}] {_bt} name={getattr(_b, 'name', '?')!r}\n")
                                elif hasattr(_b, 'content'):
                                    _f.write(f"    [{_j}] {_bt} content={getattr(_b, 'content', '')!r}\n")
                                else:
                                    _f.write(f"    [{_j}] {_bt}\n")
                            _tus = getattr(_msg, 'tool_uses', []) or []
                            if _tus:
                                _f.write(f"<<< 接收数据 - TOOL_USES ({len(_tus)}):\n")
                                for _tu in _tus:
                                    _f.write(f"    - name={_tu.name!r} id={_tu.id!r} input={_json.dumps(getattr(_tu, 'input', {}), ensure_ascii=False)!r}\n")
                            _f.write("\n")
                # --- end debug ---

                if isinstance(event, ApiTextDeltaEvent):
                    yield AssistantTextDelta(text=event.text), None
                    continue
                if isinstance(event, ApiRetryEvent):
                    yield StatusEvent(
                        message=(
                            f"Request failed; retrying in {event.delay_seconds:.1f}s "
                            f"(attempt {event.attempt + 1} of {event.max_attempts}): {event.message}"
                        )
                    ), None
                    continue

                if isinstance(event, ApiMessageCompleteEvent):
                    final_message = event.message
                    usage = event.usage
        except Exception as exc:
            error_msg = str(exc)
            if _is_completion_token_limit_error(exc):
                supported_limit = _extract_completion_token_limit(exc)
                if supported_limit is not None and effective_max_tokens > supported_limit:
                    previous_max_tokens = effective_max_tokens
                    effective_max_tokens = supported_limit
                    yield StatusEvent(
                        message=(
                            f"Model rejected max_tokens={previous_max_tokens}; "
                            f"retrying with provider limit {effective_max_tokens}."
                        )
                    ), None
                    turn_count = max(0, turn_count - 1)
                    continue
            if not reactive_compact_attempted and _is_prompt_too_long_error(exc):
                reactive_compact_attempted = True
                yield StatusEvent(message=REACTIVE_COMPACT_STATUS_MESSAGE), None
                async for event, usage in _stream_compaction(trigger="reactive", force=True):
                    yield event, usage
                compacted_messages, was_compacted = last_compaction_result
                if compacted_messages is not messages:
                    messages[:] = compacted_messages
                if was_compacted:
                    continue
            if "connect" in error_msg.lower() or "timeout" in error_msg.lower() or "network" in error_msg.lower():
                yield ErrorEvent(message=f"Network error: {error_msg}. Check your internet connection and try again."), None
            else:
                yield ErrorEvent(message=f"API error: {error_msg}"), None
            return

        if final_message is None:
            raise RuntimeError("Model stream finished without a final message")

        coordinator_context_message: ConversationMessage | None = None
        if context.system_prompt.startswith("You are a **coordinator**."):
            if messages and messages[-1].role == "user" and messages[-1].text.startswith("# Coordinator User Context"):
                coordinator_context_message = messages.pop()

        if final_message.role == "assistant" and final_message.is_effectively_empty():
            log.warning("dropping empty assistant message from provider response")
            yield ErrorEvent(
                message=(
                    "Model returned an empty assistant message. "
                    "The turn was ignored to keep the session healthy."
                )
            ), usage
            return

        messages.append(final_message)
        yield AssistantTurnComplete(message=final_message, usage=usage), usage

        if coordinator_context_message is not None:
            messages.append(coordinator_context_message)

        if not final_message.tool_uses:
            if context.hook_executor is not None:
                await context.hook_executor.execute(
                    HookEvent.STOP,
                    {
                        "event": HookEvent.STOP.value,
                        "stop_reason": "tool_uses_empty",
                    },
                )
            return

        tool_calls = final_message.tool_uses

        # --- DEBUG: log tool calls to file (接收数据 - 工具调用) ---
        if True:
            import json as _json
            with open("/home/claw/testclaude/run_query_total.txt", "a") as _f:
                _f.write(f">>> 接收数据 - TOOL_CALLS ({len(tool_calls)}):\n")
                for _tc in tool_calls:
                    _f.write(f"    - name={_tc.name!r} id={_tc.id!r}\n")
                    _f.write(f"      input={_json.dumps(getattr(_tc, 'input', {}), indent=4, ensure_ascii=False)}\n")
        # --- end debug ---

        if len(tool_calls) == 1:
            # Single tool: sequential (stream events immediately)
            tc = tool_calls[0]
            yield ToolExecutionStarted(tool_name=tc.name, tool_input=tc.input), None
            try:
                result = await _execute_tool_call(context, tc.name, tc.id, tc.input)
            except Exception as exc:
                log.exception("tool execution raised: name=%s id=%s", tc.name, tc.id)
                result = ToolResultBlock(
                    tool_use_id=tc.id,
                    content=f"Tool {tc.name} failed: {type(exc).__name__}: {exc}",
                    is_error=True,
                )
            # --- DEBUG: log single tool result (发送数据 - 工具执行结果) ---
            if True:
                with open("/home/claw/testclaude/run_query_total.txt", "a") as _f:
                    _f.write(f"<<< 发送数据 - TOOL_RESULT: name={tc.name!r} is_error={result.is_error}\n")
                    _f.write(f"    output={result.content!r}\n")
            # --- end debug ---
            yield ToolExecutionCompleted(
                tool_name=tc.name,
                output=result.content,
                is_error=result.is_error,
                metadata=result.result_metadata,
            ), None
            tool_results = [result]
        else:
            # Multiple tools: execute concurrently, emit events after
            for tc in tool_calls:
                yield ToolExecutionStarted(tool_name=tc.name, tool_input=tc.input), None

            async def _run(tc):
                return await _execute_tool_call(context, tc.name, tc.id, tc.input)

            # Use return_exceptions=True so a single failing tool does not abandon
            # its siblings as cancelled coroutines and leave the conversation with
            # un-replied tool_use blocks (Anthropic's API rejects the next request
            # on the session if any tool_use is missing a matching tool_result).
            raw_results = await asyncio.gather(
                *[_run(tc) for tc in tool_calls], return_exceptions=True
            )
            tool_results = []
            for tc, result in zip(tool_calls, raw_results):
                if isinstance(result, BaseException):
                    log.exception(
                        "tool execution raised: name=%s id=%s",
                        tc.name,
                        tc.id,
                        exc_info=result,
                    )
                    result = ToolResultBlock(
                        tool_use_id=tc.id,
                        content=f"Tool {tc.name} failed: {type(result).__name__}: {result}",
                        is_error=True,
                    )
                tool_results.append(result)

            for tc, result in zip(tool_calls, tool_results):
                # --- DEBUG: log multi tool result (发送数据 - 工具执行结果) ---
                if True:
                    with open("/home/claw/testclaude/run_query_total.txt", "a") as _f:
                        _f.write(f"<<< 发送数据 - TOOL_RESULT: name={tc.name!r} is_error={result.is_error}\n")
                        _f.write(f"    output={result.content!r}\n")
                # --- end debug ---
                yield ToolExecutionCompleted(
                    tool_name=tc.name,
                    output=result.content,
                    is_error=result.is_error,
                    metadata=result.result_metadata,
                ), None

        messages.append(ConversationMessage(role="user", content=tool_results))

    if context.max_turns is not None:
        raise MaxTurnsExceeded(context.max_turns)
    raise RuntimeError("Query loop exited without a max_turns limit or final response")

/home/claw/OpenHarness/.venv/lib/python3.11/site-packages/anyio/streams/tls.py

    async def receive(self, max_bytes: int = 65536) -> bytes:
        data = await self._call_sslobject_method(self._ssl_object.read, max_bytes)
        if not data:
            raise EndOfStream

        # --- DEBUG: log plaintext after TLS decryption ---
        import datetime as _dt_recv
        _ts = _dt_recv.datetime.now().isoformat()
        _logpath = "/home/claw/testclaude/send_receive.txt"
        with open(_logpath, "a") as _f:
            _f.write(f"\n{'─' * 80}\n")
            _f.write(f"[RECV PLAINTEXT] {_ts} len={len(data)}\n")
            _f.write(f"{'─' * 80}\n")
            _text = data.decode("utf-8", errors="replace")
            _f.write(_text)
            if not _text.endswith("\n"):
                _f.write("\n")
        # --- end debug ---

        # =================================================================
        # NEW LOG (commented out — smart file extraction + pretty-print)
        # _log_tls_plaintext("RECV", data)
        # =================================================================

        return data

    async def send(self, item: bytes) -> None:
        # --- DEBUG: log plaintext before TLS encryption ---
        import datetime as _dt_send
        _ts = _dt_send.datetime.now().isoformat()
        _logpath = "/home/claw/testclaude/send_receive.txt"
        with open(_logpath, "a") as _f:
            _f.write(f"\n{'=' * 80}\n")
            _f.write(f"[SEND PLAINTEXT] {_ts} len={len(item)}\n")
            _f.write(f"{'=' * 80}\n")
            _text = item.decode("utf-8", errors="replace")
            _f.write(_text)
            if not _text.endswith("\n"):
                _f.write("\n")
        # --- end debug ---

        # =================================================================
        # NEW LOG (commented out — smart file extraction + pretty-print)
        # _log_tls_plaintext("SEND", item)
        # =================================================================

        await self._call_sslobject_method(self._ssl_object.write, item)

4.这个是完整的openharness会话内容

> stm32f407实现跑马灯功能，然后review这个功能。

 ⏺ 
     ⏵ skill name=brainstorming → 114L

 ⏺ 
   我来帮你实现STM32F407的跑马灯功能并进行代码审查。首先让我了解一下当前项目的结构。
     ⏵ glob **/* → 1L

 ⏺ 
     ⏵ bash ls -la → 5L

 ⏺ 
     ⏵ glob docs/**/* → 1L

 ⏺ 
     ⏵ bash ls -la docs/ → 4L

 ⏺ 
     ⏵ bash ls -la docs/superpowers/ → 5L

 ⏺ 
   当前工作目录是空的，没有现有的STM32项目。让我为你创建一个完整的STM32F407跑马灯项目，然后进行代码审查。
   设计方案：
     1. 硬件选择：使用STM32F407的GPIO端口（如PD12-PD15，对应开发板上的4个LED）
     2. 实现方式：
   让我开始实现：
     ⏵ write_file path=main.c → 1L

 ⏺ 
     ⏵ write_file path=stm32f4xx_hal_conf.h → 1L

 ⏺ 
     ⏵ write_file path=Makefile → 1L

 ⏺ 
     ⏵ write_file path=STM32F407VGTx_FLASH.ld → 1L

 ⏺ 
   现在让我使用review技能对代码进行审查：
     ⏵ skill name=review → 19L

 ⏺ 
     ⏵ read_file path=main.c → 193L

 ⏺ 

   代码审查报告

   📋 项目文件结构

     ╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮
     │ main.c                    - 主程序                                                                                                                                                                                      │
     │ stm32f4xx_hal_conf.h      - HAL配置文件                                                                                                                                                                                 │
     │ Makefile                  - 构建脚本                                                                                                                                                                                    │
     │ STM32F407VGTx_FLASH.ld    - 链接器脚本                                                                                                                                                                                  │
     ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯

   ────────────────────────────────────────────────

   ✅ 优点

    ┌──────────────┬──────────────────────────────────────────────────┐
    │ 项目         │ 说明                                             │
    ├──────────────┼──────────────────────────────────────────────────┤
    │ 代码结构清晰 │ 函数职责单一，注释完整                           │
    │ 使用HAL库    │ 符合STM32现代开发规范，可移植性好                │
    │ 错误处理     │ Error_Handler 提供了错误指示机制                 │
    │ 宏定义       │ LED引脚和参数使用宏定义，便于修改                │
    │ 静态变量     │ current_led 使用 static 保持状态，作用域控制得当 │
    └──────────────┴──────────────────────────────────────────────────┘

   ────────────────────────────────────────────────

   ⚠️ 发现的问题

   🔴 Critical - 潜在严重问题
     1. main.c:180 - Error_Handler 中使用了 HAL_GPIO_TogglePin 和 HAL_Delay，但如果错误发生在时钟配置阶段（SystemClock_Config 失败），SysTick可能尚未正确配置，HAL_Delay 将无法正常工作。建议修复：

   🟡 Major - 重要改进建议
     2. main.c:82-84 - 每次循环都熄灭所有LED再点亮一个，会产生不必要的GPIO操作。可以优化为只操作变化的LED：
     3. main.c:25 - RUNNING_DELAY_MS 使用宏定义，无法在运行时调整。建议添加按键控制速度的功能，或至少提供可调接口。

   🟢 Minor -  minor改进
     4. Makefile - 依赖 STM32CubeF4 环境变量，缺少该变量时的友好提示：
     5. main.c:14-19 - LED颜色定义与Discovery开发板绑定，如果使用其他开发板需要修改。建议添加条件编译或配置注释说明。

   ────────────────────────────────────────────────

   📊 代码质量评分

    ┌──────────┬────────────┬───────────────────────────────────┐
    │ 维度     │ 评分       │ 说明                              │
    ├──────────┼────────────┼───────────────────────────────────┤
    │ 功能性   │ ⭐⭐⭐⭐⭐ │ 跑马灯功能完整实现                │
    │ 可读性   │ ⭐⭐⭐⭐⭐ │ 注释清晰，命名规范                │
    │ 可维护性 │ ⭐⭐⭐⭐☆  │ 宏定义便于修改，但缺少配置灵活性  │
    │ 健壮性   │ ⭐⭐⭐☆☆   │ Error_Handler在时钟错误时可能失效 │
    │ 效率     │ ⭐⭐⭐⭐☆  │ GPIO操作可优化                    │
    └──────────┴────────────┴───────────────────────────────────┘

   ────────────────────────────────────────────────

   📝 总结
   这是一个结构良好、功能完整的STM32F407跑马灯实现。主要问题是 Error_Handler 在时钟配置失败时可能无法正常工作，建议优先修复。其他改进建议可根据实际需求选择性采纳。
   是否需要我修复上述问题？
 ────────────────────────────────────────────────────────────
 model: kimi-k2.5 │ mode: Default
 >  
 shift+enter newline  enter send  / commands  tab mode  ↑↓ history  esc stop  ctrl+c exit

5.数据流log 和openharness生成的代码在资源里面，

OpenHarness 智能体与大模型交互的数据流中的 log 和生成的代码

https://download.csdn.net/download/weixin_43864601/93054967?spm=1001.2014.3001.5501