fix(explore): bind measurements to the declared execution result

loopx/capabilities/explore/README.md

@@ -31,6 +31,13 @@ attachments also require the same `input_revision`. Use a new finding id per
 evaluated candidate, and reuse an id only to correct that finding. Missing values
 are explicit `null`; zero and negative values remain valid measurements.
 
+When an `execution_observation` declares a `result_ref`, each measurement must
+cite that same result. This applies to manual findings and writeback attachments,
+including shared comparison contexts. Listing both receipts in `evidence_refs`
+does not bind a different result to the executed candidate. No result is inferred
+when execution has not declared one. These consistency checks do not independently
+verify a producer's calculator or turn `confirmed` into an evaluator attestation.
+
 To reuse verified comparison conditions, save these three fields once in
 `conditions.json`:
 
@@ -1413,3 +1420,17 @@ outcomes.
 Native research completion retains the graph lock and, when inheriting, the
 device-configuration lock until its transaction finishes. Inherited planning
 requires the same exact typed experiment evidence as an explicit Goal mode.
+
+Historical compatibility: stricter execution/measurement attribution is a new-write
+rule. Canonical readers preserve earlier records and their identities, including
+an inconsistent original followed by a correction. Current finding views expose
+`measurement_error` and omit contradictory quantities from comparison; raw history
+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.
+
+历史兼容：执行与测量的一致性约束限制新写入，不能阻断旧记录及其后续更正的读取。
+当前 finding 若有归因矛盾，会返回 `measurement_error` 并排除该数量比较；历史仍保留
+生产者的原始声明。正确更正可恢复当前比较，不能删改旧行。显式按顺序恢复归档前缀也
+保留此审计边界；普通单条和批量写入仍拒绝新矛盾，结构损坏、不安全及伪造记录仍拒绝。

loopx/capabilities/explore/measurements.py

@@ -21,10 +21,13 @@ def _normalize(encoded: str) -> str:
     return json.dumps(result["measurements"], sort_keys=True, allow_nan=False)
 
 
-def normalize_measurements(value, *, measurement_context=_UNSET):
+def normalize_measurements(value, *, measurement_context=_UNSET, execution_observation=_UNSET):
     validate_public_safe_value(value, path="measurements")
     request = {"measurements": value}
     if measurement_context is not _UNSET:
         validate_public_safe_value(measurement_context, path="measurement_context")
         request["measurement_context"] = measurement_context
+    if execution_observation is not _UNSET:
+        validate_public_safe_value(execution_observation, path="execution_observation")
+        request["execution_observation"] = execution_observation
     return json.loads(_normalize(json.dumps(request, sort_keys=True, allow_nan=False)))

loopx/capabilities/explore/result_log.py

@@ -352,6 +352,7 @@ def build_explore_finding_event(
     recorded_at: str | None = None,
     execution_observation: Mapping[str, Any] | None = None,
     measurements: Sequence[Mapping[str, Any]] | None = None,
+    _check_execution_measurements: bool = True,
 ) -> dict[str, Any]:
     safe_goal_id = _safe_goal_id(goal_id)
     safe_title = _compact_text(title, limit=TITLE_LIMIT, field="title")
@@ -394,7 +395,10 @@ def build_explore_finding_event(
 
         if not node_id:
             raise ValueError("measurements require an explicit node")
-        measured = normalize_measurements(measurements)
+        measured = normalize_measurements(
+            measurements, **({"execution_observation": event.get("execution_observation")}
+                if _check_execution_measurements else {}),
+        )
         if any(item["evidence_ref"] not in event.get("evidence_refs", []) for item in measured):
             raise ValueError("measurement evidence must be included in the finding's evidence_refs")
         event["measurements"] = measured
@@ -456,6 +460,8 @@ def validate_explore_result_event(
     unknown fields, invalid ids, unsafe text, forged boundary flags, and stale
     event ids. Persisted text has its own bounded compatibility budget: it is
     validated and restored before hashing, never clipped to the writer default.
+    This structural read validation preserves older attribution contradictions;
+    writers reject new contradictions, and current views exclude their numbers.
     """
 
     payload = dict(event)
@@ -518,6 +524,10 @@ def validate_explore_result_event(
             supersedes=payload.get("supersedes"),
             execution_observation=payload.get("execution_observation"),
             measurements=payload.get("measurements"),
+            # Canonical history predates the stricter new-write rule. Keep
+            # original bytes/identities auditable; current views qualify the
+            # folded result separately, and append rechecks new writes below.
+            _check_execution_measurements=False,
         )
     if summary:
         rebuilt["summary"] = summary
@@ -527,8 +537,16 @@ def validate_explore_result_event(
     return rebuilt
 
 
+def _check_finding_measurement_execution(event: Mapping[str, Any]) -> None:
+    if "measurements" in event:
+        from .measurements import normalize_measurements
+
+        normalize_measurements(event["measurements"], execution_observation=event.get("execution_observation"))
+
+
 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)
     log_path = path.expanduser()
     log_path.parent.mkdir(parents=True, exist_ok=True)
     with exclusive_file_lock(log_path):
@@ -598,6 +616,11 @@ def append_explore_result_events(
     if require_ordered_prefix and create_only_node_ids:
         raise ValueError("ordered prefix replay cannot skip existing node revisions")
     validated = [validate_explore_result_event(event, expected_goal_id=expected_goal_id) for event in events]
+    # Ordered replay restores an audited historical prefix, not new evidence.
+    # Its old contradictions remain visible only as archival facts/diagnostics.
+    if not require_ordered_prefix:
+        for event in validated:
+            _check_finding_measurement_execution(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")
@@ -841,10 +864,19 @@ def _materialized_parent_edges(
 
 
 def _finding_view(event: Mapping[str, Any]) -> dict[str, Any]:
+    measured: dict[str, Any] = {}
+    if "measurements" in event:
+        try:
+            _check_finding_measurement_execution(event)
+        except ValueError as error:
+            # Quarantine contradictory historical quantities in the current
+            # view, not the append-only history. Runtime failure propagates.
+            measured["measurement_error"] = str(error)
+        else:
+            measured.update(measurements=event["measurements"], recorded_order=event["_recorded_order"])
     return {
         **({"execution_observation": event["execution_observation"]} if "execution_observation" in event else {}),
-        **({"measurements": event["measurements"], "recorded_order": event["_recorded_order"]}
-           if "measurements" in event else {}),
+        **measured,
         "finding_id": str(event.get("result_id") or ""),
         "finding": str(event.get("title") or ""),
         "summary": str(event.get("summary") or ""),

loopx/control_plane/capabilities/explore_measurements.ts

@@ -4,6 +4,7 @@
 import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
 import {requireJsonObject, type JsonObject} from "../runtime_decode.ts";
 import {isExploreEvidenceRef} from "./explore_evidence_ref.ts";
+import {normalizeExploreExecutionObservation} from "./explore_execution_observation.ts";
 
 export interface ExploreMeasurement {
   metric: string; value: number | null; unit: string; higher_is_better: boolean;
@@ -21,7 +22,7 @@ export interface ExploreMeasurementFeedback {
   improvement: number | null; measured_count: number; unknown_count: number;
 }
 
-export function normalizeExploreMeasurements(value: unknown, context?: unknown): ExploreMeasurement[] {
+export function normalizeExploreMeasurements(value: unknown, context?: unknown, executionObservation?: unknown): ExploreMeasurement[] {
   if (!Array.isArray(value) || !value.length || value.length > 8) {
     throw new EffectRuntimeRequestError("measurements requires one to eight explicit measurements");
   }
@@ -39,10 +40,10 @@ export function normalizeExploreMeasurements(value: unknown, context?: unknown):
       // Authoring only: persist the existing complete measurement, never a
       // mutable context reference or inferred comparison membership.
       return {...shared, ...observation};
-    }));
+    }), undefined, executionObservation);
   }
   const seen = new Set<string>();
-  return value.map(item => {
+  const measurements = value.map(item => {
     const row = requireJsonObject(item, "measurement");
     const fields = ["metric", "value", "unit", "higher_is_better", "comparison_scope",
       "input_revision", "evaluation_ref", "candidate_ref", "evidence_ref"];
@@ -76,6 +77,19 @@ export function normalizeExploreMeasurements(value: unknown, context?: unknown):
     seen.add(key);
     return result;
   });
+  const mismatch = executionMeasurementError(measurements, executionObservation);
+  if (mismatch !== null) throw new EffectRuntimeRequestError(mismatch);
+  return measurements;
+}
+
+function executionMeasurementError(measurements: readonly ExploreMeasurement[], observation: unknown): string | null {
+  if (observation == null) return null;
+  const execution = normalizeExploreExecutionObservation({observation});
+  if (measurements.some(item => item.input_revision !== execution.input_revision))
+    return "measurements must retain the executed candidate's input revision";
+  if (execution.result_ref !== undefined && measurements.some(item => item.evidence_ref !== execution.result_ref))
+    return "measurements must cite the executed candidate's result reference";
+  return null;
 }
 
 function comparisonKey(node: string, metric: ExploreMeasurement): string {
@@ -101,7 +115,11 @@ export function projectExploreMeasurementFeedback(findings: readonly JsonObject[
   for (const finding of ordered) {
     const node = String(finding.node_id ?? ""), id = String(finding.finding_id ?? "");
     if (!node || !id) throw new EffectRuntimeRequestError("measurements require their node and finding identities");
-    for (const measurement of normalizeExploreMeasurements(finding.measurements)) {
+    const measurements = normalizeExploreMeasurements(finding.measurements);
+    // Old canonical history is auditable but is not a quantitative comparison
+    // source when its declared execution contradicts its measurements.
+    if (executionMeasurementError(measurements, finding.execution_observation) !== null) continue;
+    for (const measurement of measurements) {
       const key = comparisonKey(node, measurement), prior = groups.get(key);
       const previous = prior?.best ?? null;
       const point = measurement.value === null ? null : {finding_id: id,
@@ -123,5 +141,6 @@ export function projectExploreMeasurementFeedback(findings: readonly JsonObject[
 }
 
 export function normalizeExploreMeasurementsRequest(params: JsonObject): JsonObject {
-  return {measurements: normalizeExploreMeasurements(params.measurements, params.measurement_context)};
+  return {measurements: normalizeExploreMeasurements(params.measurements, params.measurement_context,
+    params.execution_observation)};
 }

loopx/control_plane/capabilities/explore_result_writeback.ts

@@ -193,7 +193,7 @@ function normalizeAttachment(value: unknown, visionPacket?: unknown, linkedScope
     throw new EffectRuntimeRequestError("measurement_context requires measurements");
   }
   const measurements = Object.hasOwn(row, "measurements")
-    ? normalizeExploreMeasurements(row.measurements, row.measurement_context) : null;
+    ? normalizeExploreMeasurements(row.measurements, row.measurement_context, execution) : null;
   if (measurements?.some(item => item.input_revision !== text(row.input_revision, "input_revision", 160)
     || !refs.includes(item.evidence_ref))) {
     throw new EffectRuntimeRequestError("measurements must retain this finding's input revision and evidence references");

loopx/control_plane/capabilities/explore_turn_context.ts

@@ -202,7 +202,7 @@ export function projectExploreTurnContext(params: JsonObject): JsonObject {
   // Manual quantitative capture uses the same finding owner. Keep its sources
   // available to the existing exact-result decision path, without requiring a
   // writeback-origin tag or admitting unrelated unmeasured notes.
-  const decisionResults = findings.filter(row => row.measurements !== undefined
+  const decisionResults = findings.filter(row => row.measurements !== undefined || row.measurement_error !== undefined
     || Array.isArray(row.tags) && row.tags.includes("writeback-result"));
   // Keep the same detail budget, but do not let unrelated newer results evict
   // the applicability of evidence linked to the next selected work item.
@@ -344,13 +344,13 @@ export function projectExploreTurnContext(params: JsonObject): JsonObject {
       counts: projection.counts ?? {},
       recent_nodes: nodes.slice(0, 3).map(row => compact(row, ["node_id", "title", "status", "blocked_reason"])),
       // The canonical evidence projection calls the finding's title `finding`.
-      recent_findings: findings.slice(0, 3).map(row => ({...compact({...row, title: row.finding}, ["finding_id", "node_id", "title", "status"]),
+      recent_findings: findings.slice(0, 3).map(row => ({...compact({...row, title: row.finding}, ["finding_id", "node_id", "title", "status", "measurement_error"]),
         ...findingDigest(row, findingSources)})),
       // Full scoped summaries; progressive reads can raise the page size.
       // These retain the observation AND applicability; clipping away conditions
       // could turn a bounded refutation into a blanket route ban.
       writeback_results: visibleResults
-        .map(row => ({...compact(row, ["finding_id", "node_id", "finding", "status", "evidence_refs"]),
+        .map(row => ({...compact(row, ["finding_id", "node_id", "finding", "status", "evidence_refs", "measurement_error"]),
           summary: String(row.summary ?? "").slice(0, EXPLORE_WRITEBACK_SUMMARY_LIMIT),
           ...(row.execution_observation ? {execution_observation: row.execution_observation} : {}),
           ...(row.measurements ? {measurements: row.measurements} : {}),

tests/control_plane/fixtures/explore_measurement_history_v0.jsonl

@@ -0,0 +1,3 @@
+{"boundary": {"absolute_paths_recorded": false, "credential_values_recorded": false, "raw_logs_recorded": false, "raw_session_transcript_recorded": false, "raw_task_text_recorded": false, "raw_trajectory_recorded": false}, "event_id": "cae05d192b3df6b2", "event_kind": "node", "goal_id": "research", "node_kind": "question", "recorded_at": "2026-01-01T00:00:00Z", "result_id": "question-a", "schema_version": "loopx_explore_result_event_v0", "status": "open", "title": "Which candidate improved?"}
+{"boundary": {"absolute_paths_recorded": false, "credential_values_recorded": false, "raw_logs_recorded": false, "raw_session_transcript_recorded": false, "raw_task_text_recorded": false, "raw_trajectory_recorded": false}, "event_id": "5bd7df96eed54e81", "event_kind": "finding", "evidence_refs": ["result:a", "result:b"], "execution_observation": {"candidate_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "change_reason": null, "elapsed_seconds": 1, "evidence_refs": ["result:a", "result:b"], "executed_method_id": "probe", "input_revision": "dataset-v1", "limit_seconds": 10, "planned_method_id": "probe", "result_candidate_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "result_ref": "result:a", "schema_version": "explore_execution_observation_v0", "search_iterations": null, "stop_reason": "process_exit:0"}, "goal_id": "research", "measurements": [{"candidate_ref": "candidate:a", "comparison_scope": "validation:budget-v1", "evaluation_ref": "evaluator:v1", "evidence_ref": "result:b", "higher_is_better": true, "input_revision": "dataset-v1", "metric": "quality", "unit": "points", "value": 8}], "node_id": "question-a", "recorded_at": "2026-01-01T01:00:00Z", "result_id": "measured", "schema_version": "loopx_explore_result_event_v0", "status": "confirmed", "title": "Measured candidate"}
+{"boundary": {"absolute_paths_recorded": false, "credential_values_recorded": false, "raw_logs_recorded": false, "raw_session_transcript_recorded": false, "raw_task_text_recorded": false, "raw_trajectory_recorded": false}, "event_id": "56e6425cc30feee1", "event_kind": "finding", "evidence_refs": ["result:a", "result:b"], "execution_observation": {"candidate_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "change_reason": null, "elapsed_seconds": 1, "evidence_refs": ["result:a", "result:b"], "executed_method_id": "probe", "input_revision": "dataset-v1", "limit_seconds": 10, "planned_method_id": "probe", "result_candidate_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "result_ref": "result:a", "schema_version": "explore_execution_observation_v0", "search_iterations": null, "stop_reason": "process_exit:0"}, "goal_id": "research", "measurements": [{"candidate_ref": "candidate:a", "comparison_scope": "validation:budget-v1", "evaluation_ref": "evaluator:v1", "evidence_ref": "result:a", "higher_is_better": true, "input_revision": "dataset-v1", "metric": "quality", "unit": "points", "value": 8}], "node_id": "question-a", "recorded_at": "2026-01-01T02:00:00Z", "result_id": "measured", "schema_version": "loopx_explore_result_event_v0", "status": "confirmed", "title": "Measured candidate"}