From 90d7d460b010777b583bec0ab8327a843cfa0202 Mon Sep 17 00:00:00 2001 From: song Date: Sun, 1 Nov 2026 10:56:35 +0800 Subject: [PATCH] fix(explore): preflight mixed source history recovery Signed-off-by: song --- loopx/capabilities/explore/README.md | 24 ++ loopx/capabilities/explore/result_log.py | 9 +- .../explore/source_history_reconcile.py | 47 +++- ...est_explore_source_history_measurements.py | 255 ++++++++++++++++++ 4 files changed, 326 insertions(+), 9 deletions(-) create mode 100644 tests/control_plane/test_explore_source_history_measurements.py diff --git a/loopx/capabilities/explore/README.md b/loopx/capabilities/explore/README.md index dcfc1d4b8e..3c78b9a65c 100644 --- a/loopx/capabilities/explore/README.md +++ b/loopx/capabilities/explore/README.md @@ -1430,7 +1430,31 @@ without deleting its prior version. Explicit ordered-prefix archival replay keep the same audit boundary; ordinary single/batch writes still reject contradictions. Malformed, unsafe, unknown-field or forged canonical records remain rejected. +`explore source-history-reconcile` merges the latest missing registered results; +it does not restore a complete archival prefix. Preview and execution apply the +same ordinary-write validation to that subset. Eligible results can be restored +even when other historical findings contradict their execution. The plan reports +`blocked_event_count`, a digest of the blocked event ids and `blocked_reasons`; +warnings also show the reasons in Markdown output. Blocked records remain in the +unchanged candidate source, do not enter quantitative comparison and do not by +themselves require deleting registered sink rows. An all-blocked subset reports +`blocked`; a partial recovery reports `would_reconcile_with_blocked_history` or +`reconciled_with_blocked_history`. Append a valid correction for the same result +id to the candidate history and preview again; recovery selects that latest +record without rewriting its older evidence. Unrelated remote orphans retain +their existing classification and status suffix. + 历史兼容：执行与测量的一致性约束限制新写入，不能阻断旧记录及其后续更正的读取。 当前 finding 若有归因矛盾，会返回 `measurement_error` 并排除该数量比较；历史仍保留 生产者的原始声明。正确更正可恢复当前比较，不能删改旧行。显式按顺序恢复归档前缀也 保留此审计边界；普通单条和批量写入仍拒绝新矛盾，结构损坏、不安全及伪造记录仍拒绝。 + +`explore source-history-reconcile` 合并已注册且缺失的各结果最新记录，不是恢复完整 +归档前缀。预览与执行对该子集应用相同的普通写入校验；部分旧 finding 归因矛盾不会 +阻断其他合法结果恢复。计划通过 `blocked_event_count`、被阻断事件 id 的摘要和 +`blocked_reasons` 报告原因，Markdown 的 warnings 同时显示这些原因。被阻断记录 +保留在未改动的候选源中，不进入数量比较，也不因此要求删除已注册的 sink 行。 +全部阻断时状态为 `blocked`，部分恢复时为 `would_reconcile_with_blocked_history` +或 `reconciled_with_blocked_history`。在候选历史中为同一 result id 追加正确更正后 +重新预览；恢复选择该最新记录，不能覆写旧证据。其他远端孤儿仍保留原有分类与状态 +后缀。 diff --git a/loopx/capabilities/explore/result_log.py b/loopx/capabilities/explore/result_log.py index 8647f42e27..3412f53922 100644 --- a/loopx/capabilities/explore/result_log.py +++ b/loopx/capabilities/explore/result_log.py @@ -537,7 +537,8 @@ def validate_explore_result_event( return rebuilt -def _check_finding_measurement_execution(event: Mapping[str, Any]) -> None: +def validate_explore_result_event_attribution(event: Mapping[str, Any]) -> None: + """Check typed measurement attribution after canonical event validation.""" if "measurements" in event: from .measurements import normalize_measurements @@ -546,7 +547,7 @@ def _check_finding_measurement_execution(event: Mapping[str, Any]) -> None: def append_explore_result_event(path: Path, event: Mapping[str, Any]) -> dict[str, Any]: validated = validate_explore_result_event(event) - _check_finding_measurement_execution(validated) + validate_explore_result_event_attribution(validated) log_path = path.expanduser() log_path.parent.mkdir(parents=True, exist_ok=True) with exclusive_file_lock(log_path): @@ -620,7 +621,7 @@ def append_explore_result_events( # Its old contradictions remain visible only as archival facts/diagnostics. if not require_ordered_prefix: for event in validated: - _check_finding_measurement_execution(event) + validate_explore_result_event_attribution(event) requested_ids = [str(event["event_id"]) for event in validated] if len(requested_ids) != len(set(requested_ids)): raise ValueError("Explore result event batch contains duplicate event ids") @@ -867,7 +868,7 @@ def _finding_view(event: Mapping[str, Any]) -> dict[str, Any]: measured: dict[str, Any] = {} if "measurements" in event: try: - _check_finding_measurement_execution(event) + validate_explore_result_event_attribution(event) except ValueError as error: # Quarantine contradictory historical quantities in the current # view, not the append-only history. Runtime failure propagates. diff --git a/loopx/capabilities/explore/source_history_reconcile.py b/loopx/capabilities/explore/source_history_reconcile.py index 7846c3d264..87abca4985 100644 --- a/loopx/capabilities/explore/source_history_reconcile.py +++ b/loopx/capabilities/explore/source_history_reconcile.py @@ -17,6 +17,7 @@ append_explore_result_events, build_explore_result_projection, load_explore_result_events_strict, + validate_explore_result_event_attribution, ) @@ -113,6 +114,9 @@ def reconcile_explore_source_history( source is eligible. Its direct projection key must already exist in the registered sink baseline. Derived rows that remain absent are classified, never synthesized or deleted by this non-destructive command. + This selected subset is an ordinary append, not an archival prefix replay: + contradictory latest findings remain in the candidate source and are + reported as blocked, while eligible results can still be recovered. """ canonical_events = load_explore_result_events_strict(canonical_log_path, goal_id=goal_id) @@ -133,18 +137,35 @@ def reconcile_explore_source_history( canonical_result_ids = {str(event.get("result_id") or "") for event in canonical_events} candidate_latest = _latest_by_result_id(candidate_events) - selected_events = [ + missing_events = [ event for result_id, event in candidate_latest.items() if result_id not in canonical_result_ids and _direct_projection_key(event, goal_id=goal_id) in missing_before ] - selected_events.sort( + missing_events.sort( key=lambda event: ( str(event.get("recorded_at") or ""), str(event.get("event_id") or ""), ) ) + selected_events = [] + blocked_event_ids: set[str] = set() + blocked_keys: set[str] = set() + blocked_reasons: Counter[str] = Counter() + for event in missing_events: + try: + validate_explore_result_event_attribution(event) + except ValueError as error: + # Both complete sources already passed structural validation. + # Reuse the writer's stricter eligibility check for this subset; + # runtime failures must propagate rather than look like bad data. + blocked_event_ids.add(str(event["event_id"])) + blocked_keys.add(_direct_projection_key(event, goal_id=goal_id)) + blocked_reasons[str(error)] += 1 + else: + selected_events.append(event) + planned_events = [*canonical_events, *selected_events] planned_projection = _projection(planned_events, goal_id=goal_id) planned_keys = _projection_result_keys(planned_projection, goal_id=goal_id) @@ -156,7 +177,7 @@ def reconcile_explore_source_history( recovered = missing_before & planned_keys remaining = registered - planned_keys stale_parent_edges = remaining & _materialized_parent_edge_keys(candidate_projection, goal_id=goal_id) - unresolved_orphans = remaining - stale_parent_edges + unresolved_orphans = remaining - stale_parent_edges - blocked_keys selected_event_ids = {str(event.get("event_id") or "") for event in selected_events} writeback = { @@ -197,11 +218,22 @@ def reconcile_explore_source_history( effective_remaining = registered - effective_keys if selected_events: status = "reconciled" if execute else "would_reconcile" + if blocked_event_ids: + status += "_with_blocked_history" + elif blocked_event_ids: + status = "blocked" else: status = "already_reconciled" - if effective_remaining: + if effective_remaining - blocked_keys: status += "_with_remote_orphans" + warnings = [ + f"Recovery excludes {count} historical result(s): {reason}" + for reason, count in sorted(blocked_reasons.items()) + ] + if warnings: + warnings.append("Append a corrected latest record to the candidate history, then preview recovery again.") + return { "ok": True, "schema_version": SCHEMA_VERSION, @@ -221,6 +253,9 @@ def reconcile_explore_source_history( "plan": { "selected_event_count": len(selected_events), "selected_event_ids_sha256_16": _digest(selected_event_ids), + "blocked_event_count": len(blocked_event_ids), + "blocked_event_ids_sha256_16": _digest(blocked_event_ids), + "blocked_reasons": dict(sorted(blocked_reasons.items())), "latest_event_per_missing_result": True, "raw_history_copied": False, }, @@ -240,12 +275,14 @@ def reconcile_explore_source_history( }, "classification": { "lost_history_recovered_count": len(recovered), + "blocked_history_count": len(blocked_keys), "stale_materialized_parent_edge_count": len(stale_parent_edges), "unresolved_orphan_count": len(unresolved_orphans), - "remote_deletion_required": bool(effective_remaining), + "remote_deletion_required": bool(effective_remaining - blocked_keys), "remote_deletion_performed": False, }, "writeback": writeback, + **({"warnings": warnings} if warnings else {}), "boundary": { **PUBLIC_BOUNDARY, "candidate_events_strictly_validated": True, diff --git a/tests/control_plane/test_explore_source_history_measurements.py b/tests/control_plane/test_explore_source_history_measurements.py new file mode 100644 index 0000000000..5aaa6659f5 --- /dev/null +++ b/tests/control_plane/test_explore_source_history_measurements.py @@ -0,0 +1,255 @@ +from __future__ import annotations + +from copy import deepcopy +import hashlib +import json +from pathlib import Path +import subprocess +import sys + +import pytest + +from loopx.capabilities.explore.result_log import ( + append_explore_result_event, + build_explore_result_projection, + explore_result_log_path, + load_explore_result_events_strict, +) +from loopx.capabilities.explore.source_history_reconcile import reconcile_explore_source_history +from loopx.control_plane.effect_runtime import effect_runtime_result + + +FIXTURE = Path(__file__).with_name("fixtures") / "explore_measurement_history_v0.jsonl" +GOAL = "research" +RESULT_REASON = "measurements must cite the executed candidate's result reference" +INPUT_REASON = "measurements must retain the executed candidate's input revision" + + +def _rehash(event): + stable = {key: value for key, value in event.items() if key != "event_id"} + event["event_id"] = hashlib.sha256( + json.dumps(stable, sort_keys=True, ensure_ascii=True).encode() + ).hexdigest()[:16] + return event + + +def _encoded(events): + return "".join(json.dumps(event, sort_keys=True) + "\n" for event in events).encode() + + +def _history(tmp_path): + # The existing fixture was written under the earlier canonical contract. + node, mixed, corrected = [json.loads(line) for line in FIXTURE.read_text().splitlines()] + valid = _rehash({**deepcopy(corrected), "result_id": "valid"}) + mixed = deepcopy(mixed) + mixed["result_id"] = "mixed-result" + mixed["measurements"][0]["value"] = 99 + _rehash(mixed) + other_input = deepcopy(corrected) + other_input["result_id"] = "mixed-input" + other_input["measurements"][0].update(input_revision="dataset-other", value=500) + _rehash(other_input) + events = [node, valid, mixed, other_input] + canonical = explore_result_log_path(tmp_path / "runtime", GOAL) + candidate = explore_result_log_path(tmp_path / "candidate", GOAL) + for path, rows in ((canonical, [node]), (candidate, events)): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(_encoded(rows)) + keys = {f"{GOAL}:nodes:question-a"} | { + f"{GOAL}:findings:{event['result_id']}" for event in events[1:] + } + return canonical, candidate, events, keys + + +def _reconcile(canonical, candidate, registered, *, execute=False): + return reconcile_explore_source_history( + canonical_log_path=canonical, candidate_log_path=candidate, + goal_id=GOAL, registered_result_keys=registered, execute=execute, + ) + + +@pytest.mark.parametrize("include_valid", [True, False]) +def test_mixed_history_recovers_eligible_subset_and_reports_blocked_reasons(tmp_path, include_valid): + canonical, candidate, events, registered = _history(tmp_path) + if not include_valid: + registered.remove(f"{GOAL}:findings:valid") + before = canonical.read_bytes() + original = candidate.read_bytes() + + preview = _reconcile(canonical, candidate, registered) + assert canonical.read_bytes() == before + assert preview["plan"]["selected_event_count"] == int(include_valid) + assert preview["plan"]["blocked_event_count"] == 2 + assert preview["plan"]["blocked_reasons"] == {RESULT_REASON: 1, INPUT_REASON: 1} + assert preview["classification"]["blocked_history_count"] == 2 + assert preview["classification"]["unresolved_orphan_count"] == 0 + assert preview["classification"]["remote_deletion_required"] is False + assert preview["projection_reconciliation"]["remaining_registered_result_count"] == 2 + assert preview["status"] == ("would_reconcile_with_blocked_history" if include_valid else "blocked") + + executed = _reconcile(canonical, candidate, registered, execute=True) + assert executed["plan"] == preview["plan"] + assert executed["classification"] == preview["classification"] + assert executed["status"] == ("reconciled_with_blocked_history" if include_valid else "blocked") + assert executed["writeback"]["appended_event_count"] == int(include_valid) + restored = load_explore_result_events_strict(canonical, goal_id=GOAL) + assert restored == (events[:2] if include_valid else events[:1]) + assert canonical.read_bytes().startswith(before) + assert candidate.read_bytes() == original + # Only admitted measurements reach the real typed feedback owner. + projection = build_explore_result_projection(restored, goal_id=GOAL) + graph = effect_runtime_result("explore.turn_context", dict( + goal_id=GOAL, agent_id="worker", route=["loopx"], graph_enabled=True, + harness_gate={"enabled": True}, projection=projection, plan={}, + ))["graph"] + feedback = graph.get("measurement_feedback", []) + assert len(feedback) == int(include_valid) + if include_valid: + assert feedback[0]["best"]["value"] == 8 + assert feedback[0]["measured_count"] == 1 + + recovered = canonical.read_bytes() + repeated = _reconcile(canonical, candidate, registered, execute=True) + assert repeated["status"] == "blocked" + assert repeated["plan"]["selected_event_count"] == 0 + assert repeated["plan"]["blocked_reasons"] == preview["plan"]["blocked_reasons"] + assert repeated["writeback"]["performed"] is False + assert canonical.read_bytes() == recovered + assert candidate.read_bytes() == original + + +def test_blocked_history_and_unrelated_remote_orphans_remain_distinct(tmp_path): + canonical, candidate, _, registered = _history(tmp_path) + registered.add(f"{GOAL}:findings:missing-source") + preview = _reconcile(canonical, candidate, registered) + assert preview["status"] == "would_reconcile_with_blocked_history_with_remote_orphans" + assert preview["classification"]["blocked_history_count"] == 2 + assert preview["classification"]["unresolved_orphan_count"] == 1 + assert preview["classification"]["remote_deletion_required"] is True + executed = _reconcile(canonical, candidate, registered, execute=True) + assert executed["classification"] == preview["classification"] + assert executed["writeback"]["appended_event_count"] == 1 + + +def test_corrected_latest_history_recovers_without_rewriting_older_evidence(tmp_path): + canonical, candidate, events, registered = _history(tmp_path) + _reconcile(canonical, candidate, registered, execute=True) + original = candidate.read_bytes() + corrections = [] + for event in events[2:]: + corrected = deepcopy(event) + corrected["recorded_at"] = "2026-01-02T00:00:00Z" + execution = corrected["execution_observation"] + corrected["measurements"][0].update( + evidence_ref=execution["result_ref"], input_revision=execution["input_revision"], + ) + corrections.append(_rehash(corrected)) + append_explore_result_event(candidate, corrected) + corrected_source = candidate.read_bytes() + assert corrected_source.startswith(original) + + preview = _reconcile(canonical, candidate, registered) + assert preview["plan"]["selected_event_count"] == 2 + assert preview["plan"]["blocked_event_count"] == 0 + executed = _reconcile(canonical, candidate, registered, execute=True) + assert executed["plan"] == preview["plan"] + assert executed["status"] == "reconciled" + assert executed["writeback"]["appended_event_count"] == 2 + restored = load_explore_result_events_strict(canonical, goal_id=GOAL) + assert {row["event_id"] for row in restored} == { + row["event_id"] for row in [*events[:2], *corrections] + } + before = canonical.read_bytes() + repeated = _reconcile(canonical, candidate, registered, execute=True) + assert repeated["status"] == "already_reconciled" + assert repeated["writeback"]["performed"] is False + assert canonical.read_bytes() == before + assert candidate.read_bytes() == corrected_source + + +@pytest.mark.parametrize("damage", ["forged-id", "unknown-field", "private-text"]) +def test_invalid_source_is_rejected_before_recovering_any_subset(tmp_path, damage): + canonical, candidate, events, registered = _history(tmp_path) + invalid = events[-1] + if damage == "forged-id": + invalid["event_id"] = "forged" + elif damage == "unknown-field": + invalid["raw_transcript"] = "unsupported" + _rehash(invalid) + else: + invalid["summary"] = "to" + "ken=" + "synthetic-value" + _rehash(invalid) + candidate.write_bytes(_encoded(events)) + before, original = canonical.read_bytes(), candidate.read_bytes() + for execute in (False, True): + with pytest.raises(ValueError): + _reconcile(canonical, candidate, registered, execute=execute) + assert canonical.read_bytes() == before + assert candidate.read_bytes() == original + + +def test_runtime_failure_is_not_misreported_as_blocked_history(tmp_path, monkeypatch): + from loopx.capabilities.explore import source_history_reconcile + + canonical, candidate, _, registered = _history(tmp_path) + before = canonical.read_bytes() + + def unavailable(*args, **kwargs): + raise RuntimeError("measurement runtime unavailable") + + monkeypatch.setattr(source_history_reconcile, "validate_explore_result_event_attribution", unavailable) + for execute in (False, True): + with pytest.raises(RuntimeError, match="runtime unavailable"): + _reconcile(canonical, candidate, registered, execute=execute) + assert canonical.read_bytes() == before + + +def test_real_cli_mixed_history_preview_execute_and_readback(tmp_path): + from loopx.extensions import lark + from loopx.extensions.runtime import default_extension_state_file, install_extension + from tests.capabilities.test_explore_turn_context import registry + + registry_path = registry(tmp_path, graph=True, planning=True) + canonical, candidate, _, registered = _history(tmp_path) + original, before = candidate.read_bytes(), canonical.read_bytes() + config = tmp_path / "board.json" + config.write_text(json.dumps({"result_records": {key: "synthetic-row" for key in registered}})) + command = [ + sys.executable, "-m", "loopx.cli", "--registry", str(registry_path), + "--runtime-root", str(tmp_path / "runtime"), "explore", "source-history-reconcile", + "--goal-id", GOAL, "--candidate-runtime-root", str(tmp_path / "candidate"), + "--config-path", str(config), + ] + + def run(*args): + return subprocess.run([*command, *args], capture_output=True, text=True, timeout=60) + + denied = run("--format", "json", "--execute") + assert denied.returncode != 0 + assert "not installed" in denied.stdout + assert canonical.read_bytes() == before + installed = install_extension( + Path(lark.__file__).with_name("extension.toml"), + state_file=default_extension_state_file(tmp_path / "runtime"), execute=True, + ) + assert installed["doctor"]["verified"] is True + preview_result = run("--format", "json") + assert preview_result.returncode == 0, preview_result.stdout + preview = json.loads(preview_result.stdout) + assert preview["plan"]["selected_event_count"] == 1 + assert preview["plan"]["blocked_reasons"] == {RESULT_REASON: 1, INPUT_REASON: 1} + markdown = run("--format", "markdown") + assert markdown.returncode == 0, markdown.stdout + assert RESULT_REASON in markdown.stdout and INPUT_REASON in markdown.stdout + assert canonical.read_bytes() == before + executed_result = run("--format", "json", "--execute") + assert executed_result.returncode == 0, executed_result.stdout + executed = json.loads(executed_result.stdout) + assert executed["plan"] == preview["plan"] + assert executed["writeback"]["appended_event_count"] == 1 + assert executed["writeback"]["readback_verified"] is True + assert {row["result_id"] for row in load_explore_result_events_strict(canonical, goal_id=GOAL)} == { + "question-a", "valid", + } + assert candidate.read_bytes() == original + assert str(tmp_path) not in preview_result.stdout