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

apps/presentation/dashboard/smoke/explore-finding-diagnostics-smoke.ts

@@ -0,0 +1,31 @@
+import assert from "node:assert/strict";
+import {fetchExploreResults} from "../src/data/chat";
+
+// The saved-finding transport used to strip the backend's quarantine reason.
+// A measured, quarantined and unmeasured finding must remain distinguishable.
+const row = {finding_id: "legacy", finding: "Recorded finding", summary: "Original observation",
+  status: "confirmed", node_id: "question", question: "Compare candidates", scope: "Fixed input",
+  agent_id: "recorder", evidence_refs: ["result:original"], last_updated_at: "2026-01-01T00:00:00Z",
+  linked_todos: []};
+const measurement = {metric: "quality", value: 8, unit: "points", higher_is_better: true,
+  comparison_scope: "validation:fixed", input_revision: "input:v1", evaluation_ref: "method:v1",
+  candidate_ref: "candidate:a", evidence_ref: "result:original"};
+const reason = "measurements must cite the executed candidate's result reference";
+const previousFetch = globalThis.fetch;
+try {
+  globalThis.fetch = async () => new Response(JSON.stringify({ok: true, goal_id: "research", total: 3,
+    next_cursor: null, items: [{...row, measurement_error: reason},
+      {...row, finding_id: "measured", measurements: [measurement]}, {...row, finding_id: "unmeasured"}]}),
+    {headers: {"Content-Type": "application/json"}});
+  const page = await fetchExploreResults("research");
+  assert.equal(page.items[0]!.measurement_error, reason, "Quarantine reason reaches the existing reader");
+  assert.equal(page.items[0]!.measurements, undefined, "Quarantined quantities stay absent");
+  assert.equal(page.items[0]!.status, "confirmed", "The transport does not rewrite the recorded finding status");
+  assert.equal(page.items[1]!.measurements?.[0]?.value, 8, "Valid quantities retain their value");
+  assert.equal(page.items[1]!.measurement_error, undefined);
+  assert.equal(page.items[2]!.measurements, undefined);
+  assert.equal(page.items[2]!.measurement_error, undefined, "Unmeasured is not a quarantine error");
+} finally {
+  globalThis.fetch = previousFetch;
+}
+console.log("PASS saved finding measurement diagnostics, valid values and unmeasured parity");

apps/presentation/dashboard/src/data/chat.ts

@@ -2914,7 +2914,8 @@ const exploreResultPageSchema = z.object({
     finding_id: z.string(), finding: z.string(), summary: z.string(), status: z.string(),
     node_id: z.string(), question: z.string(), scope: z.string(), agent_id: z.string(),
     evidence_refs: z.array(z.string()), last_updated_at: z.string(),
-    measurements: z.unknown().transform(normalizeExploreMeasurements).optional(),
+    measurements: z.unknown().transform(value => normalizeExploreMeasurements(value)).optional(),
+    measurement_error: z.string().optional(),
     linked_todos_available: z.boolean().optional(),
     linked_todos: z.array(z.object({
       todo_id: z.string(), text: z.string(), status: z.string(), claimed_by: z.string(),

apps/presentation/dashboard/src/features/personal-workspace/goal-explore-results.css

@@ -17,6 +17,7 @@
 .goal-explore-results .goal-team-result-reader { overflow-wrap: anywhere; }
 .goal-explore-results .goal-team-result-reader > h4 { font-size: var(--text-lead); line-height: var(--leading-lead); font-weight: 600; margin: 0 0 12px; }
 .explore-reader-status { color: var(--pw-muted); font-size: var(--text-caption); }
+.explore-measurement-warning { margin-block: 16px; font-size: var(--text-body); line-height: var(--leading-body); }
 .goal-explore-results .goal-team-result-reader p, .goal-explore-results .goal-team-result-reader li { font-size: var(--text-lead); line-height: var(--leading-prose); color: var(--pw-text); }
 .explore-finding-prose, .explore-finding-raw { max-width: 72ch; }
 .explore-finding-prose section { margin: 24px 0; }

apps/presentation/dashboard/src/features/personal-workspace/goal-explore-results.tsx

@@ -1,6 +1,6 @@
 import {exploreFindingCopy} from "./explore-finding-copy";
 import {useEffect, useRef, useState} from "react";
-import {ChevronRight, RefreshCw} from "lucide-react";
+import {AlertTriangle, ChevronRight, RefreshCw} from "lucide-react";
 import {fetchExploreResults, readTodoRequest, type ExploreResultPage} from "../../data/chat";
 import {parseExploreFindingSections} from "../../../../../../loopx/control_plane/presentation/explore_timeline.js";
 import {ExploreMeasurements} from "./explore-measurements";
@@ -111,7 +111,13 @@ export function GoalExploreResults({goalId, zh, active = true, readSourceUrl}: {
       {selected ? <div ref={reader} tabIndex={-1} className="goal-team-result-reader" key={selected.finding_id}>
         <h4>{selected.question || selected.finding}</h4>
         <span className="explore-reader-status">{statuses[selected.status] ?? selected.status} · <time dateTime={selected.last_updated_at}>{time(selected.last_updated_at)}</time></span>
-        {selected.measurements ? <ExploreMeasurements measurements={selected.measurements} zh={zh}/> : null}
+        {selected.measurement_error ? <div className="explore-reader-heading explore-measurement-warning" role="note">
+          <span>{zh ? "测量归属有冲突，未参与比较。" : "Measurements excluded: conflicting attribution."}</span>
+          <ContextHelp icon={AlertTriangle} label={zh ? "测量冲突与更正" : "Measurement conflict and correction"}>
+            <p>{selected.measurement_error}</p>
+            <p>{zh ? "请记录者追加执行与测量归属一致的更正结论；原记录仍保留在历史中。" : "Ask the recorder to append a corrected finding with matching execution and measurement references. The original stays in history."}</p>
+          </ContextHelp>
+        </div> : selected.measurements ? <ExploreMeasurements measurements={selected.measurements} zh={zh}/> : null}
         {sections ? <div className="explore-finding-prose">
           <section><h5>{zh ? "观察" : "Observation"}</h5><p>{sections.observation}</p></section>
           {sections.interpretation && !sections.decision ? <section><h5>{zh ? "解读" : "Interpretation"}</h5><p>{sections.interpretation}</p></section> : null}

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,35 @@ 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.
+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
+ordinary write/ownership checks, and conflicting attachment retries reject.
+CLI attachment parsing preserves structurally valid historical input until the
+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.
+
+历史兼容：执行与测量的一致性约束限制新写入，不能阻断旧记录及其后续更正的读取。
+当前 finding 若有归因矛盾，会返回 `measurement_error` 并排除该数量比较；历史仍保留
+生产者的原始声明。正确更正可恢复当前比较，不能删改旧行。显式按顺序恢复归档前缀也
+保留此审计边界；普通单条和批量写入仍拒绝新矛盾，结构损坏、不安全及伪造记录仍拒绝。
+已登记来源的恢复对选中的缺失结果使用相同结构校验，在锁内绑定 canonical 快照，
+保持预览与执行一致；不能把这一子集称为完整归档前缀。已完成交付仅重放原始持久化
+ingestion intent，并精确匹配来源、记录时间、canonical finding 与既有 Todo 关联；
+此路径不写图或 Todo。未完成交付仍须遵守普通写入与归属检查，冲突附件重试仍拒绝。
+CLI 附件解析保留结构合法的历史输入，由既有 primary owner 精确匹配不可变重试；
+新 primary 写入和新 ingestion intent 仍在产生效果前检查执行与测量归属。
+解析成功不代表新写入合格，也不提供公开的绕过参数。

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,35 @@ 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)
+    return _append_validated_explore_result_events(
+        path, validated, expected_goal_id=expected_goal_id,
+        create_only_node_ids=create_only_node_ids,
+        require_ordered_prefix=require_ordered_prefix,
+        expected_event_ids=expected_event_ids,
+    )
+
+
+def _append_validated_explore_result_events(
+    path: Path,
+    validated: Sequence[dict[str, Any]],
+    *,
+    expected_goal_id: str,
+    create_only_node_ids: Sequence[str] = (),
+    require_ordered_prefix: bool = False,
+    expected_event_ids: Sequence[str] | None = None,
+) -> 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.
+    """
     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")
@@ -616,6 +663,11 @@ def append_explore_result_events(
             if current_events[:len(validated)] != validated[:len(current_events)]:
                 raise ValueError("Explore result history differs from the required initial prefix")
             existing_by_id = {str(row["event_id"]): row for row in current_events}
+        elif expected_event_ids is not None:
+            existing_by_id = {
+                str(row["event_id"]): row for row in
+                load_explore_result_events_strict(log_path, goal_id=expected_goal_id)
+            }
         elif log_path.exists():
             for line in log_path.read_text(encoding="utf-8").split("\n"):
                 try:
@@ -841,10 +893,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/capabilities/explore/result_writeback.py

@@ -98,7 +98,8 @@ def _evolution_plan(attachment, projection, *, agent_id):
     })
 
 
-def _events(attachment, *, goal_id, agent_id, source_id, recorded_at, existing_node=None):
+def _events(attachment, *, goal_id, agent_id, source_id, recorded_at, existing_node=None,
+            _check_execution_measurements=True):
     previous = existing_node or {}
     update = attachment.get("question_update") or {}
     node = build_explore_node_event(
@@ -135,6 +136,7 @@ def _events(attachment, *, goal_id, agent_id, source_id, recorded_at, existing_n
         tags=["writeback-result"],
         execution_observation=attachment.get("execution_observation"),
         measurements=attachment.get("measurements"),
+        _check_execution_measurements=_check_execution_measurements,
         recorded_at=recorded_at,
     )
     relations = [build_explore_edge_event(
@@ -307,6 +309,11 @@ def deliver_result_attachment(
         if len(intents) != 1:
             raise ValueError("Explore result hook did not admit one ingestion intent")
         intent = intents[0]
+        # Exact old payloads remain readable; equivalent current shorthand
+        # still normalizes through the typed owner before comparing intents.
+        if (attachment != intent["payload"]["attachment"]
+                and normalize_result_attachment(attachment) != intent["payload"]["attachment"]):
+            raise ValueError("Explore attachment conflicts with its committed ingestion intent")
         # Readback-only replay after success needs no new lease or Todo write.
         existing_events = load_explore_result_events_strict(
             explore_result_log_path(runtime_root, goal_id), goal_id=goal_id
@@ -317,6 +324,12 @@ def deliver_result_attachment(
             agent_id=agent_id,
             source_id=intent["idempotency_key"],
             recorded_at=committed_at,
+            # A durable terminal intent can describe an old writer's event.
+            # Structural reconstruction is only for exact canonical readback;
+            # a missing event still goes through strict preparation below.
+            _check_execution_measurements=(
+                "explore.result_writeback" not in dispatch.get("replayed_hooks", [])
+            ),
         )
         detail = list_goal_todos(
             registry_path=registry_path,