diff --git a/loopx/capabilities/explore/README.md b/loopx/capabilities/explore/README.md
index 759a057d22..8af07a525e 100644
--- a/loopx/capabilities/explore/README.md
+++ b/loopx/capabilities/explore/README.md
@@ -1429,9 +1429,6 @@ retains the producer's original statement. A matching correction becomes usable
 without deleting its prior version. Explicit ordered-prefix archival replay keeps
 the same audit boundary; ordinary single/batch writes still reject contradictions.
 Malformed, unsafe, unknown-field or forged canonical records remain rejected.
-Registered-source recovery applies the same structural validation to its selected
-missing results, binds execution to the canonical snapshot under lock, and keeps
-preview and execution consistent. It does not label that subset an archival prefix.
 Completed delivery replays only the original durable ingestion intent, exact
 source and recorded time, matching canonical finding and existing Todo link.
 That path performs no graph or Todo mutation; missing delivery still requires the
@@ -1441,14 +1438,39 @@ existing primary owner matches the immutable retry. New primary writes and new
 ingestion intents still validate execution/measurement attribution before effects;
 parsing is not a new-write eligibility decision or a public bypass.
 
+`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; execution binds to the canonical
+snapshot under lock. 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` 并排除该数量比较；历史仍保留
 生产者的原始声明。正确更正可恢复当前比较，不能删改旧行。显式按顺序恢复归档前缀也
 保留此审计边界；普通单条和批量写入仍拒绝新矛盾，结构损坏、不安全及伪造记录仍拒绝。
-已登记来源的恢复对选中的缺失结果使用相同结构校验，在锁内绑定 canonical 快照，
-保持预览与执行一致；不能把这一子集称为完整归档前缀。已完成交付仅重放原始持久化
+已完成交付仅重放原始持久化
 ingestion intent，并精确匹配来源、记录时间、canonical finding 与既有 Todo 关联；
 此路径不写图或 Todo。未完成交付仍须遵守普通写入与归属检查，冲突附件重试仍拒绝。
 CLI 附件解析保留结构合法的历史输入，由既有 primary owner 精确匹配不可变重试；
 新 primary 写入和新 ingestion intent 仍在产生效果前检查执行与测量归属。
 解析成功不代表新写入合格，也不提供公开的绕过参数。
+
+`explore source-history-reconcile` 合并已注册且缺失的各结果最新记录，不是恢复完整
+归档前缀。预览与执行对该子集应用相同的普通写入校验，并在锁内绑定 canonical
+快照；部分旧 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 5400950124..9d9deef6c4 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)
     return _append_validated_explore_result_events(
         path, validated, expected_goal_id=expected_goal_id,
         create_only_node_ids=create_only_node_ids,
@@ -640,10 +641,11 @@ def _append_validated_explore_result_events(
 ) -> dict[str, Any]:
     """Shared file effect after the owning caller qualifies the events.
 
-    Ordinary writes validate new attribution above. Source reconciliation may
-    supply structurally validated history only after checking its registered
-    baseline, missing result ids and exact canonical snapshot. This internal
-    seam adds no public validation switch or historical-prefix claim.
+    Ordinary writes validate new attribution above. Source reconciliation
+    preflights the same attribution after checking its registered baseline and
+    missing result ids, then binds execution to the exact canonical snapshot.
+    This internal seam adds no public validation switch or historical-prefix
+    claim.
     """
     requested_ids = [str(event["event_id"]) for event in validated]
     if len(requested_ids) != len(set(requested_ids)):
@@ -896,7 +898,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 1475cdc7d1..8eaccdbe2b 100644
--- a/loopx/capabilities/explore/source_history_reconcile.py
+++ b/loopx/capabilities/explore/source_history_reconcile.py
@@ -17,6 +17,7 @@
     _append_validated_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 = {
@@ -168,9 +189,9 @@ def reconcile_explore_source_history(
     }
     post_keys = canonical_keys
     if execute and selected_events:
-        # These are existing facts from the validated source, selected only for
-        # registered missing ids. They are neither new evidence nor an ordered
-        # archival prefix. Bind this plan to the canonical snapshot under lock.
+        # The registered missing subset passed structural and attribution
+        # preflight; it is not an archival prefix. Bind this plan to the
+        # canonical snapshot under lock.
         append_receipt = _append_validated_explore_result_events(
             canonical_log_path,
             selected_events,
@@ -201,11 +222,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,
@@ -225,6 +257,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,
         },
@@ -244,12 +279,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_measurement_history.py b/tests/control_plane/test_explore_measurement_history.py
index fa93e70c4e..69a87a34fd 100644
--- a/tests/control_plane/test_explore_measurement_history.py
+++ b/tests/control_plane/test_explore_measurement_history.py
@@ -92,7 +92,7 @@ def test_historical_read_still_rejects_forged_canonical_records(tmp_path):
         load_explore_result_events_strict(log, goal_id="research")
 
 
-def test_registered_history_preview_executes_without_promoting_legacy_measurements(tmp_path):
+def test_registered_history_recovers_eligible_subset_without_promoting_legacy_measurements(tmp_path):
     from loopx.capabilities.explore.source_history_reconcile import reconcile_explore_source_history
 
     # The real writer at 21367015 produced these bytes, including two old
@@ -109,16 +109,20 @@ def test_registered_history_preview_executes_without_promoting_legacy_measuremen
         registered_result_keys={"research:nodes:question-a", "research:findings:valid",
                                 "research:findings:mixed", "research:findings:mixed-later"})
     preview = reconcile_explore_source_history(**args)
-    assert preview["status"] == "would_reconcile"
-    assert preview["projection_reconciliation"]["recovered_lost_history_count"] == 3
+    assert preview["status"] == "would_reconcile_with_blocked_history"
+    assert preview["projection_reconciliation"]["recovered_lost_history_count"] == 1
+    assert preview["plan"]["blocked_event_count"] == 2
     assert canonical.read_bytes() == prefix
     receipt = reconcile_explore_source_history(**args, execute=True)
-    assert receipt["status"] == "reconciled"
-    assert receipt["writeback"]["appended_event_count"] == 3
+    assert receipt["status"] == "reconciled_with_blocked_history"
+    assert receipt["writeback"]["appended_event_count"] == 1
     assert receipt["writeback"]["readback_verified"] is True
-    assert canonical.read_bytes() == candidate.read_bytes() == original
+    restored = canonical.read_bytes()
+    assert load_explore_result_events_strict(canonical, goal_id="research") == events[:2]
+    assert restored.startswith(prefix)
+    assert candidate.read_bytes() == original
     repeated = reconcile_explore_source_history(**args, execute=True)
-    assert repeated["status"] == "already_reconciled"
+    assert repeated["status"] == "blocked"
     assert repeated["writeback"]["performed"] is False
     projection = build_explore_result_projection(events, goal_id="research")
     assert sum("measurement_error" in row for row in projection["findings"]) == 2
@@ -129,14 +133,14 @@ def test_registered_history_preview_executes_without_promoting_legacy_measuremen
     assert feedback[0]["best"]["value"] == 8
     with pytest.raises(ValueError, match="executed candidate's result"):
         append_explore_result_events(canonical, events[1:], expected_goal_id="research")
-    assert canonical.read_bytes() == original
+    assert canonical.read_bytes() == restored
     corrected = {key: events[2][key] for key in ("goal_id", "title", "node_id", "status",
         "evidence_refs", "execution_observation")}
     correction = build_explore_finding_event(**corrected, finding_id="mixed",
         recorded_at="2026-01-01T04:00:00Z", measurements=[{**events[2]["measurements"][0],
             "value": 7, "evidence_ref": "result:a"}])
     append_explore_result_event(canonical, correction)
-    assert canonical.read_bytes().startswith(original)
+    assert canonical.read_bytes().startswith(restored)
     request["projection"] = build_explore_result_projection(
         load_explore_result_events_strict(canonical, goal_id="research"), goal_id="research")
     group = effect_runtime_result("explore.turn_context", request)["graph"]["measurement_feedback"][0]
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