diff --git a/app/harness/observation.py b/app/harness/observation.py index b1b9d83..c8e8b5a 100644 --- a/app/harness/observation.py +++ b/app/harness/observation.py @@ -348,6 +348,9 @@ def write_gate_evidence( question_id 字段承载 **unit_id**(single=question_id,pair=pair_id)—— 逐题明细在 predictions 表溯源,按 pair_id join 真实 question 表会 join 不上。 + 返回: + 无。 + 关键实现: 逐行 insert(非 insert_many),保证每行独立事务。 """ diff --git a/app/harness/validate.py b/app/harness/validate.py index d16dbe6..b2228c0 100644 --- a/app/harness/validate.py +++ b/app/harness/validate.py @@ -1246,9 +1246,12 @@ async def validate_skills_concurrent( ) -> dict[str, ValidationOutcome]: """连续并发 gate:多题型全部臂共享题槽并发,统计按阶梯序前缀有序推进。 - 发射顺序 = 题型 round-robin × 题型内阶梯序(base 先 cand 后);题型过线即 - 冻结,其排队任务启动时自查冻结标志撤销,in-flight 结果不计入(τ 之后样本, - 合法丢弃)。全部题型判定后统一组装 ValidationOutcome。 + 关键实现细节: + 发射顺序 = 题型 round-robin × 题型内阶梯序(base 先 cand 后);题型过线 + 即冻结,其排队任务启动时自查冻结标志撤销,in-flight 结果不计入(τ 之后 + 样本,合法丢弃);候选目录逐个物化即登记、统一 finally 清理(中途失败不 + 泄漏);任一任务异常先 cancel+排水其余任务再向上传播;全部题型判定后 + 统一经 _finalize_outcome 组装。 参数: workspace_dir: workspace 根目录(候选物化用)。 diff --git a/tests/unit/test_gate_concurrent.py b/tests/unit/test_gate_concurrent.py index 38acd42..cad1bda 100644 --- a/tests/unit/test_gate_concurrent.py +++ b/tests/unit/test_gate_concurrent.py @@ -16,6 +16,14 @@ from tests.unit.test_gate_prefix import _PARAMS, _mk_unit from tests.unit.test_gate_unit_arm import _FakeLog +class _FakeInferenceResult: + """推理结果桩:只承载编排器消费的 run_id 与 total 两个字段。""" + + def __init__(self, run_id: str, total: int) -> None: + self.run_id = run_id + self.total = total + + def _mk_spec(task_type: str, slug: str, n: int) -> GateSpec: """构造 n 个 single 单元的 gate 规格(unit_id 形如 -q)。""" return GateSpec( @@ -31,11 +39,6 @@ def _mk_spec(task_type: str, slug: str, n: int) -> GateSpec: def _scripted_inference(log: _FakeLog, script: dict[str, tuple[bool, float]]): """脚本化假推理:按 question_id+臂 决定 (对错, 延迟秒),制造乱序到达。""" - class _R: - def __init__(self, run_id: str, total: int) -> None: - self.run_id = run_id - self.total = total - async def _run(questions, *, run_id: str, skills_dir: Path): arm = "cand" if run_id.endswith("_cand") else "base" correct, delay = script[f"{questions[0].question_id}|{arm}"] @@ -51,7 +54,7 @@ def _scripted_inference(log: _FakeLog, script: dict[str, tuple[bool, float]]): "steps_json": "[]", } ) - return _R(run_id, len(questions)) + return _FakeInferenceResult(run_id, len(questions)) return _run @@ -130,10 +133,6 @@ async def test_all_infra_raises(tmp_path, monkeypatch) -> None: spec = _mk_spec("Action Reasoning", "action-reasoning", 2) log = _FakeLog() - class _R: - def __init__(self, run_id, total): - self.run_id, self.total = run_id, total - async def _infra_run(questions, *, run_id, skills_dir): for q in questions: log.rows.append( @@ -146,7 +145,7 @@ async def test_all_infra_raises(tmp_path, monkeypatch) -> None: "steps_json": "[]", } ) - return _R(run_id, len(questions)) + return _FakeInferenceResult(run_id, len(questions)) monkeypatch.setattr( "app.harness.validate.materialize_candidate_skill", @@ -217,10 +216,6 @@ async def test_guard_raise_cancels_remaining_tasks(tmp_path, monkeypatch) -> Non log = _FakeLog() hang = asyncio.Event() # 永不 set:B 型推理只能靠取消收束 - class _R: - def __init__(self, run_id, total): - self.run_id, self.total = run_id, total - async def _run(questions, *, run_id, skills_dir): if "counting-problem" in run_id: await hang.wait() @@ -235,7 +230,7 @@ async def test_guard_raise_cancels_remaining_tasks(tmp_path, monkeypatch) -> Non "steps_json": "[]", } ) - return _R(run_id, len(questions)) + return _FakeInferenceResult(run_id, len(questions)) monkeypatch.setattr( "app.harness.validate.materialize_candidate_skill",