Skip to content

fix(explore): preflight mixed source history recovery for #7764 #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
songoow wants to merge 1 commit into from
Closed

fix(explore): preflight mixed source history recovery for #7764 #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions loopx/capabilities/explore/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 追加正确更正后
重新预览；恢复选择该最新记录，不能覆写旧证据。其他远端孤儿仍保留原有分类与状态
后缀。
9 changes: 5 additions & 4 deletions loopx/capabilities/explore/result_log.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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):
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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.
Expand Down
47 changes: 42 additions & 5 deletions loopx/capabilities/explore/source_history_reconcile.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
append_explore_result_events,
build_explore_result_projection,
load_explore_result_events_strict,
validate_explore_result_event_attribution,
)


Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand All @@ -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 = {
Expand Down Expand Up @@ -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,
Expand All @@ -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,
},
Expand All @@ -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,
Expand Down
Loading