diff --git a/app/harness/inference.py b/app/harness/inference.py index 42f58c9..a9a6136 100644 --- a/app/harness/inference.py +++ b/app/harness/inference.py @@ -409,7 +409,11 @@ async def _run_single_question( 返回: 预测结果字典（含 video_id, question_id, prediction, answer 等）。 """ + # run_id 必须显式入 record:HarnessLog.insert 缺省用**实例** run_id 填充, + # 连续并发 gate 共享单一 gate_log(实例 run_id 为 step 级)时,各臂行必须 + # 落自己的臂 run_id,否则 validate 回读 _load_run_rows(臂 run_id) 为空。 record: dict[str, Any] = { + "run_id": run_id, "video_id": qa.video_id, "question_id": qa.question_id, "task_type": qa.task_type, diff --git a/app/harness/runner.py b/app/harness/runner.py index 013ecb2..01c0c89 100644 --- a/app/harness/runner.py +++ b/app/harness/runner.py @@ -1370,13 +1370,17 @@ class Runner: rejected=state.rejected_buffer.get(task_type, []), ) - records = dict( - zip( - active_types, - await asyncio.gather(*[_evolve_one(t) for t in active_types]), - strict=True, - ) - ) + # 首异常先取消其余进化任务并排水再向上传播(与 validate_skills_concurrent + # 同款语义):避免失败后残留 in-flight LLM 任务与 pending task 警告。 + tasks = [asyncio.ensure_future(_evolve_one(t)) for t in active_types] + try: + evolved = await asyncio.gather(*tasks) + except BaseException: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + raise + records = dict(zip(active_types, evolved, strict=True)) # 无真实改动的题型照旧写 skipped 后出队 gated: dict[str, EvolutionRecord] = {} diff --git a/tests/unit/test_gate_batch_parallel.py b/tests/unit/test_gate_batch_parallel.py index 6ff1725..2708e36 100644 --- a/tests/unit/test_gate_batch_parallel.py +++ b/tests/unit/test_gate_batch_parallel.py @@ -236,3 +236,65 @@ def test_gate_batch_parallel_evolve_and_alphabetical_settle( assert "_gate_" in spec.gate_run_prefix assert isinstance(captured["log"], _FakeHarnessLog) assert callable(captured["run_inference"]) + + +# --------------------------------------------------------------------------- +# 共享 gate_log 的 run_id 契约(真 SQLite,Codex 质量审 C1): +# HarnessLog.insert 缺省用实例 run_id 填充;record 自带 run_id 必须覆盖它, +# 否则连续并发 gate 下所有臂的 predictions 会落成 step 级 run_id, +# validate 按臂 run_id 回读为空 → gate 静默废掉。 +# --------------------------------------------------------------------------- + + +def test_harness_log_insert_record_run_id_overrides_instance(tmp_path: Path) -> None: + """record 自带 run_id 覆盖实例 run_id;缺省时回落实例 run_id(锁死 enriched.update 语义)。""" + from app.harness.inference import PREDICTIONS_SCHEMA + from app.harness.log import HarnessLog + + with HarnessLog(str(tmp_path / "harness.db"), "gate_e1_s0") as log: + log.create_table("predictions", PREDICTIONS_SCHEMA) + log.insert( + "predictions", + {"run_id": "run_e1_s0_gate_a_base_u0", "question_id": "q1", "prediction": "A"}, + ) + log.insert("predictions", {"question_id": "q2", "prediction": "B"}) + rows = log.query("SELECT question_id, run_id FROM predictions ORDER BY question_id") + assert [(r["question_id"], r["run_id"]) for r in rows] == [ + ("q1", "run_e1_s0_gate_a_base_u0"), + ("q2", "gate_e1_s0"), + ] + + +def test_inference_prediction_row_carries_arm_run_id(tmp_path: Path) -> None: + """经共享 gate_log 落库的 prediction 行 run_id 必须是臂 run_id 而非实例 run_id。 + + prompt_builder 抛错走异常路径即落库,无需真实 LLM; + 该路径与成功路径共用同一 record 初始 dict,契约一致。 + """ + from app.harness.inference import PREDICTIONS_SCHEMA, _run_single_question + from app.harness.log import HarnessLog + + def _broken_prompt_builder(qa: GeneratedQuestion) -> tuple[str, str]: + raise RuntimeError("测试注入:跳过真实推理") + + async def _noop_dispatch(tool_name: str, args: dict, *, context: dict) -> str: + raise NotImplementedError + + with HarnessLog(str(tmp_path / "harness.db"), "gate_e1_s0") as gate_log: + gate_log.create_table("predictions", PREDICTIONS_SCHEMA) + asyncio.run( + _run_single_question( + _question("q-arm", _TYPE_A), + llm=object(), # prompt_builder 先抛错,不会触达 + tool_dispatch_fn=_noop_dispatch, + prompt_builder=_broken_prompt_builder, + log=gate_log, + max_steps=3, + plugins=[], + run_id="run_e1_s0_gate_action-reasoning_cand_u0", + ) + ) + rows = gate_log.query("SELECT run_id, stop_reason FROM predictions") + assert len(rows) == 1 + assert rows[0]["run_id"] == "run_e1_s0_gate_action-reasoning_cand_u0" + assert rows[0]["stop_reason"] == "error" diff --git a/tests/unit/test_harness_inference.py b/tests/unit/test_harness_inference.py index 5481cb9..c00346c 100644 --- a/tests/unit/test_harness_inference.py +++ b/tests/unit/test_harness_inference.py @@ -117,8 +117,9 @@ def harness_log(tmp_path: Any, request: Any) -> HarnessLog: """创建临时 HarnessLog 实例。 使用 test 节点名称的 hash 作为 db 文件名，避免冲突。 - run_id 固定为 "test-run"，实际 run_inference 中传入的 run_id - 由 HarnessLog.insert 自动覆盖为 HarnessLog 构造时的值。 + 实例 run_id 固定为 "test-run"；predictions 行的 run_id 由 inference + record 显式携带（run_inference 传入值），不回落实例 run_id—— + 连续并发 gate 共享单一 HarnessLog 的契约。 """ db_name = f"harness_{id(request)}.db" db_path = str(tmp_path / db_name) @@ -528,8 +529,9 @@ class TestPredictionAlwaysWritten: assert result.correct == 0 assert result.stop_reason_counts.get("error") == 1 - # 验证 DB 中的记录（HarnessLog.insert 使用构造时的 run_id） - rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",)) + # 验证 DB 中的记录（record 显式携带 run_inference 的 run_id， + # 不再回落 HarnessLog 实例 run_id——连续并发 gate 共享 log 的契约） + rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-error",)) assert len(rows) == 1 assert rows[0]["stop_reason"] == "error" assert rows[0]["prediction"] is None @@ -553,8 +555,8 @@ class TestPredictionAlwaysWritten: ) assert result.total == 1 - # HarnessLog.insert 使用构造时的 run_id - rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",)) + # record 显式携带 run_inference 的 run_id（共享 log 契约） + rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-parse-err",)) assert len(rows) == 1 assert rows[0]["prediction"] is None @@ -612,7 +614,7 @@ class TestNonScalarPrediction: ) assert result.total == 1 - rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",)) + rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-nonscalar",)) assert len(rows) == 1 # prediction 被 JSON 序列化为字符串，不再是 Python list assert rows[0]["prediction"] == '["B"]' diff --git a/tests/unit/test_inference_pair_aggregate.py b/tests/unit/test_inference_pair_aggregate.py index 7d07aa8..e806eda 100644 --- a/tests/unit/test_inference_pair_aggregate.py +++ b/tests/unit/test_inference_pair_aggregate.py @@ -332,8 +332,8 @@ class TestRunInferencePairEndToEnd: assert result.total == 1 assert result.correct == 1 - # 逐题溯源：predictions 表两条 record 都在 - rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",)) + # 逐题溯源：predictions 表两条 record 都在（record 显式携带传入的 run_id） + rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-pair-e2e",)) qids = {r["question_id"] for r in rows} assert qids == {"po", "pm"} @@ -360,6 +360,6 @@ class TestRunInferencePairEndToEnd: ) assert result.total == 1 # single 存活，孤儿剔除 - # 逐题溯源：孤儿题仍逐题落库（推理不变） - rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",)) + # 逐题溯源：孤儿题仍逐题落库（推理不变；record 显式携带传入的 run_id） + rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-orphan-e2e",)) assert {r["question_id"] for r in rows} == {"s1", "po"}