From 01f2e7c7b9301e5cd61dbda951d73d49cac29344 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 15 Jul 2026 11:51:55 -0400 Subject: [PATCH] feat: add steps_json-backed RunLog wrapper --- app/harness/baseline_run_log.py | 71 +++++++++++++++++++++++++++++ tests/unit/test_baseline_run_log.py | 33 ++++++++++++++ 2 files changed, 104 insertions(+) create mode 100644 app/harness/baseline_run_log.py create mode 100644 tests/unit/test_baseline_run_log.py diff --git a/app/harness/baseline_run_log.py b/app/harness/baseline_run_log.py new file mode 100644 index 0000000..beaae39 --- /dev/null +++ b/app/harness/baseline_run_log.py @@ -0,0 +1,71 @@ +"""RunLog 包装器:traces 表空时从 predictions.steps_json 重建轨迹。 + +用于对 infer_adhoc 这类 traces 未落表、轨迹在 steps_json 的历史 run 跑离线诊断。 +""" + +from __future__ import annotations + +from typing import Any + +from app.harness.steps_json_traces import steps_json_to_trace_rows + + +class StepsJsonRunLog: + """委托内层 RunLog;get_traces 空表时回退 steps_json。 + + 实现 core/evolution/protocols.py 的 RunLog Protocol(duck-typing)。 + 对 traces 已落表的正常 run 完全透传;仅当底层 traces 为空时, + 才从 predictions.steps_json 经 steps_json_to_trace_rows 重建轨迹行。 + """ + + def __init__(self, inner: Any) -> None: + """构造包装器。 + + 参数: + inner: 内层 RunLog 实现(如 app/harness/log.py::RunLogImpl), + 需提供 get_predictions / get_traces 两个 async 方法。 + """ + self._inner = inner + + async def get_predictions( + self, run_id: str, *, question_ids: list[str] | None = None + ) -> list[dict[str, Any]]: + """透传内层预测查询。 + + 参数: + run_id: 运行标识。 + question_ids: 可选的题目 ID 过滤列表。 + + 返回: + 内层返回的预测记录字典列表,原样透传。 + """ + return await self._inner.get_predictions(run_id, question_ids=question_ids) + + async def get_traces( + self, run_id: str, *, question_ids: list[str] | None = None + ) -> list[dict[str, Any]]: + """查询轨迹;底层 traces 表空时从 steps_json 回退重建。 + + 参数: + run_id: 运行标识。 + question_ids: 可选的题目 ID 过滤列表。 + + 返回: + 轨迹行字典列表。 + + 关键实现细节: + - 内层 traces 非空 → 原样返回,不触发回退(正常 run 路径)。 + - 内层 traces 为空 → 拉取同一过滤条件下的 predictions, + 逐题经 steps_json_to_trace_rows 展开为轨迹行并拼接。 + - steps_json 缺失时以空串传入,由下游确定性返回 []。 + """ + inner_rows = await self._inner.get_traces(run_id, question_ids=question_ids) + if inner_rows: + return inner_rows + preds = await self._inner.get_predictions(run_id, question_ids=question_ids) + rows: list[dict[str, Any]] = [] + for p in preds: + rows.extend( + steps_json_to_trace_rows(p["video_id"], p["question_id"], p.get("steps_json") or "") + ) + return rows diff --git a/tests/unit/test_baseline_run_log.py b/tests/unit/test_baseline_run_log.py new file mode 100644 index 0000000..f6b1ce9 --- /dev/null +++ b/tests/unit/test_baseline_run_log.py @@ -0,0 +1,33 @@ +import json + +import pytest + +from app.harness.baseline_run_log import StepsJsonRunLog + + +class _FakeInner: + def __init__(self, preds, traces): + self._preds, self._traces = preds, traces + + async def get_predictions(self, run_id, *, question_ids=None): + return [p for p in self._preds if not question_ids or p["question_id"] in question_ids] + + async def get_traces(self, run_id, *, question_ids=None): + return list(self._traces) + + +@pytest.mark.asyncio +async def test_get_traces_falls_back_to_steps_json_when_table_empty(): + steps = [{"thought": "t", "tool_call": {"tool": "view_node", "args": {}}, "tool_output": "o"}] + preds = [{"video_id": "v1", "question_id": "q1", "steps_json": json.dumps(steps)}] + log = StepsJsonRunLog(_FakeInner(preds, traces=[])) + rows = await log.get_traces("r", question_ids=["q1"]) + assert rows[0]["tool_name"] == "view_node" and rows[0]["question_id"] == "q1" + + +@pytest.mark.asyncio +async def test_get_traces_prefers_nonempty_inner_table(): + inner_traces = [{"video_id": "v1", "question_id": "q1", "step": 0, "tool_name": "x"}] + log = StepsJsonRunLog(_FakeInner([], inner_traces)) + rows = await log.get_traces("r") + assert rows == inner_traces