diff --git a/loopx/capabilities/explore/result_log.py b/loopx/capabilities/explore/result_log.py index 9d9deef6c4..4f5eaa4380 100644 --- a/loopx/capabilities/explore/result_log.py +++ b/loopx/capabilities/explore/result_log.py @@ -1157,11 +1157,14 @@ def build_explore_result_projection( _edge_view(event) for event in _fold_by_result_id(scoped, event_kind=EVENT_KIND_EDGE) ] edges = [*edges, *_materialized_parent_edges(nodes, edges)] - findings = sorted( - (_finding_view(event) for event in folded_findings), - key=lambda item: chronology_key(item["last_updated_at"]), + # Whole-second source clocks can tie. Folding retains creation order, so + # break equal chronology keys with the existing latest-append ordinal + # before bounding any consumer; do not rewrite the stored evidence clock. + findings = [_finding_view(event) for event in sorted( + folded_findings, + key=lambda item: (chronology_key(item["last_updated_at"]), item["_recorded_order"]), reverse=True, - ) + )] nodes_by_status: dict[str, int] = {status: 0 for status in sorted(NODE_STATUSES)} for node in nodes: diff --git a/tests/control_plane/test_explore_measurements.py b/tests/control_plane/test_explore_measurements.py index cc5f5f74a1..d2a7eac762 100644 --- a/tests/control_plane/test_explore_measurements.py +++ b/tests/control_plane/test_explore_measurements.py @@ -8,6 +8,8 @@ from tests.control_plane.test_explore_result_writeback import fixture, payload from loopx.capabilities.explore.result_log import ( + build_explore_finding_event, + build_explore_result_projection, explore_history_rows, explore_result_log_path, load_explore_result_events_strict, @@ -29,6 +31,36 @@ def context(args): ("registry_path", "runtime_root", "goal_id", "agent_id")}) +@pytest.mark.parametrize("measured", [False, True]) +def test_tied_finding_times_bound_latest_appends_and_corrections(measured): + # The production writer records whole seconds. Equal source times must not + # let insertion order keep the oldest result in a bounded newest-first view. + stamp = "2026-01-01T00:00:00Z" + + def finding(index, *, at=stamp, **changes): + return build_explore_finding_event( + goal_id="research", finding_id=f"candidate-{index}", node_id="question", + title=f"Measured scope {index}", evidence_refs=["validation:counterexample-1"], + tags=["writeback-result"] if not measured else [], recorded_at=at, + **({"measurements": [measurement(str(index), index)]} if measured else {}), **changes, + ) + + events = [finding(index) for index in range(4)] + initial = build_explore_result_projection(events, goal_id="research", finding_limit=3) + assert [row["finding_id"] for row in initial["findings"]] == ["candidate-3", "candidate-2", "candidate-1"] + events.append(finding(0, summary="Corrected interpretation on the same source clock.")) + original = deepcopy(events) + view = build_explore_result_projection(events, goal_id="research", finding_limit=3) + assert [row["finding_id"] for row in view["findings"]] == ["candidate-0", "candidate-3", "candidate-2"] + assert view["findings"][0]["update_count"] == 2 + assert all(row["last_updated_at"] == stamp for row in view["findings"]) + # Preserve chronology across distinct timestamps, even for a later append. + older = finding(4, at="2025-12-31T23:59:59Z") + dated = build_explore_result_projection([*events, older], goal_id="research", finding_limit=5) + assert dated["findings"][-1]["finding_id"] == "candidate-4" + assert events == original, "ordering is a projection; source timestamps and append history stay unchanged" + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_cli_reuses_explicit_context_and_preserves_normalized_history(tmp_path, monkeypatch, provider): from canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture @@ -67,6 +99,11 @@ def test_cli_reuses_explicit_context_and_preserves_normalized_history(tmp_path, assert feedback["best"]["value"] == 9 assert feedback["measured_count"] == 3 assert feedback["unknown_count"] == 1 + # Actual File/SQLite CLI writes may share the writer's whole-second clock. + # The bounded reader must agree with their append order without sleeps. + assert [row["finding_id"] for row in context(args)["graph"]["writeback_results"]] == [ + "candidate-3", "candidate-2", "candidate-1", + ] assert conditions.read_text() == json.dumps(shared), "recording cannot rewrite the caller's conditions" healthy = log.read_bytes() # Even an equal duplicate field must not silently override the shared source.