fix(explore): preflight mixed source history recovery

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 追加正确更正后
+重新预览；恢复选择该最新记录，不能覆写旧证据。其他远端孤儿仍保留原有分类与状态
+后缀。

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.

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,

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]