diff --git a/apps/presentation/dashboard/src/features/personal-workspace/goal-research-history.tsx b/apps/presentation/dashboard/src/features/personal-workspace/goal-research-history.tsx
index e06ec8d728..6d616d40b2 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/goal-research-history.tsx
+++ b/apps/presentation/dashboard/src/features/personal-workspace/goal-research-history.tsx
@@ -75,9 +75,10 @@ export function GoalResearchHistory({goalId, active, zh, refreshRevision = 0, re
   const outcomeCopy: Record<string, string> = zh ? {replan: "转向", continue: "继续", stop: "停止", wait: "等待", no_change: "无变化", ask_human: "请人决定"} : {replan: "Replan", continue: "Continue", stop: "Stop", wait: "Wait", no_change: "No change", ask_human: "Ask human"};
   const findingBody = (finding: ExploreHistoryEvent) => {
     const sections = parseExploreFindingSections(finding.summary);
-    if (!sections) return <p>{finding.summary}</p>;
-    const decision = sections.decision;
-    return <dl className="research-finding-sections">
+    const decision = sections?.decision;
+    const execution = finding.execution_observation;
+    const measured = (value: unknown) => value === null ? (zh ? "未测量" : "Not measured") : String(value);
+    return <>{sections ? <dl className="research-finding-sections">
       <dt>{sectionCopy.observation}</dt><dd>{sections.observation}</dd>
       {decision ? <>
         <dt>{zh ? "决策" : "Decision"}</dt><dd><b className={`research-outcome is-${decision.outcome}`}>{outcomeCopy[decision.outcome] ?? decision.outcome}</b></dd>
@@ -85,7 +86,17 @@ export function GoalResearchHistory({goalId, active, zh, refreshRevision = 0, re
       </> : <><dt>{sectionCopy.interpretation}</dt><dd>{sections.interpretation}</dd></>}
       <dt>{sectionCopy.applicability}</dt><dd className="is-muted">{sections.applicability}</dd>
       <dt>{sectionCopy.inputRevision}</dt><dd><code>{sections.inputRevision}</code></dd>
-    </dl>;
+    </dl> : <p>{finding.summary}</p>}{execution ? <details><summary>{zh ? "实际执行与资源观测" : "Execution and resource observations"}</summary>
+      <dl className="research-finding-sections">
+        <dt>{zh ? "计划方法 → 实际方法" : "Planned → executed method"}</dt><dd>{String(execution.planned_method_id)} → {String(execution.executed_method_id)}</dd>
+        {execution.change_reason ? <><dt>{zh ? "改向原因" : "Reason for change"}</dt><dd>{String(execution.change_reason)}</dd></> : null}
+        <dt>{zh ? "运行秒数 / 上限秒数" : "Elapsed / limit seconds"}</dt><dd>{measured(execution.elapsed_seconds)} / {measured(execution.limit_seconds)}</dd>
+        <dt>{zh ? "搜索次数" : "Search iterations"}</dt><dd>{measured(execution.search_iterations)}</dd>
+        <dt>{zh ? "结束原因" : "Termination reason"}</dt><dd>{measured(execution.stop_reason)}</dd>
+        <dt>{zh ? "执行候选 SHA-256" : "Executed candidate SHA-256"}</dt><dd><code>{String(execution.candidate_sha256)}</code></dd>
+        {execution.result_ref ? <><dt>{zh ? "该候选的结果引用" : "Result for this candidate"}</dt><dd><code>{String(execution.result_ref)}</code></dd></> : null}
+      </dl><p>{zh ? "来源记录的执行观测；不等于方法有效或当前候选已通过验收。" : "Source-recorded execution observations; not proof of method benefit or current-candidate acceptance."}</p>
+    </details> : null}</>;
   };
   const findingElement = (finding: ExploreHistoryEvent) => <div className="research-recorded-finding" key={finding.event_id}><strong>{finding.title}</strong><small>{timeLabel(finding.recorded_at)} · {zh ? "来源标记" : "Source label"} · {stateLabel(finding.status)}{finding.confidence !== null ? ` · ${zh ? "来源自报置信度" : "Source-reported confidence"} ${finding.confidence}` : ""}</small>{findingBody(finding)}{finding.evidence_refs.map(ref => <code key={ref}>{ref}</code>)}</div>;
   const updateRecords = (events: readonly ExploreHistoryEvent[]) => <ol className="research-history-records">{events.map(event => <li key={event.event_id}><time>{timeLabel(event.recorded_at)}</time><span>{stateLabel(event.status)} · {event.title}{event.summary ? ` — ${event.summary}` : ""}</span>{event.parent_id ? <code>{copy.group}: {event.parent_id}</code> : null}{event.evidence_refs.map(ref => <code key={ref}>{ref}</code>)}{event.run_id ? <code>{event.run_id}</code> : null}</li>)}</ol>;
diff --git a/benchmark/runtime/RESEARCH_LOOP.md b/benchmark/runtime/RESEARCH_LOOP.md
new file mode 100644
index 0000000000..ffad446133
--- /dev/null
+++ b/benchmark/runtime/RESEARCH_LOOP.md
@@ -0,0 +1,74 @@
+# Research stages and execution evidence
+
+The heartbeat runtime's explicit `research_loop=True` option retains the
+original research objective while separating an initial-candidate Todo from
+its deferred research successor. Omission preserves the previous treatment.
+The runtime selects the existing `evidence_conditioned_v1` replan strategy and
+rejects an explicitly conflicting strategy. This standalone stage/evidence
+entry retains the existing Host admission boundaries; it does not install a
+per-provider-request guard. When composed with C1's native request-admission
+entry, that entry checks ordinary Host admission at each actual provider send
+and requires a configured Responses-compatible endpoint.
+
+The entry creates the two stages using existing `todo_done` and successor
+semantics. Finishing the candidate releases real optimization work; it cannot
+by itself close that work. Research retains the best candidate, meaningful
+negative results and reasoned closeout against the original objective. User
+Stop, the original budget and genuine blockers remain authoritative.
+Preparation records scope through the ordinary Agent Vision writer as
+`surface_only`, without claiming research progress or spending a work Turn.
+Existing research work is not reopened or its later Vision overwritten.
+
+The installed `loopx-research-probe` uses the staged runner/Python. The same
+producer can be called as `python -m benchmark.runtime.research_probe`:
+
+```sh
+loopx-research-probe --candidate-file candidate.py --output probe-001.json \
+  --input-revision public-case-v1 --planned-method-id scheduler \
+  --executed-method-id mask-selection --change-reason 'The control isolated mask selection first.' \
+  --evidence-ref probe:local-001 --limit-seconds 30 --diagnostics probe-diagnostics.json \
+  -- python3 candidate.py
+```
+
+The solver may write `{"search_iterations":2,"stop_reason":"search_cap"}` to
+`LOOPX_RESEARCH_DIAGNOSTICS`. The host measures elapsed time, exit/timeout and
+the selected candidate-file digest before/after. A changed candidate rejects
+attribution. Solver counters and reasons remain reports; missing counters
+remain null. Low resource use does not impose a mandatory time floor.
+Malformed optional diagnostics preserve reliable host measurements and retain
+a bounded diagnostic-error receipt. Evidence paths must be fresh; raw output
+stays local. The probe neither submits nor accesses hidden evaluation.
+
+The adjacent `.candidate.json` preserves selected relative paths and hashes.
+`candidate_sha256` hashes the sorted JSON file manifest, not native archive
+container bytes. Pins cover only declared files at the probe boundaries, not
+transitive imports or independent writers. Result attribution must still prove
+that the evaluated snapshot matches the selected candidate and input revision.
+
+Attach the emitted `execution_observation` through normal Explore result
+writeback. Its input revision and evidence refs must belong to the enclosing
+finding; released results must identify the candidate actually evaluated.
+Different planned/executed method ids require a change reason. Findings,
+replan and research-history details retain this same source. They do not
+certify method benefit or Goal completion. Missing observations preserve old
+events; a reader for the optional field is required for new retained findings.
+
+Result followthrough uses existing material delivery classifications.
+`surface_only` and unknown historical outcomes remain readable but cannot
+consume a cited work result or create a new unresolved experiment. Useful
+negative results and concrete blockers remain eligible evidence.
+
+## 中文交付边界
+
+显式研究入口保留原 Goal，把初始候选与后续研究表示成既有 Todo 和后继依赖。
+候选完成只释放后续研究，准备阶段使用 `surface_only`，不会声称已经取得收益。
+Stop、原预算和真实阻塞保持有效，不要求填满时长或改变原生评测规则。
+独立的阶段/证据入口沿用既有 Host 准入边界，不安装逐次 provider 请求守卫。
+只有组合 C1 的原生请求准入入口后，该入口才会在每次实际 provider 发送前检查
+普通 Host 准入，并要求配置兼容 Responses 的端点。
+
+本地探针固定所选文件，记录真实耗时、退出与超时状态；可选 solver 报告与宿主
+测量分开，坏诊断不能抹去可靠测量。所选文件清单哈希不是原生归档哈希。
+证据通过既有 Explore 写回并进入 replan 和研究历程；缺失测量保持未知，负结果
+继续可用。该切片交付本地执行证据与阶段接续，原生提交、自动反馈依赖和成本
+归约由后续组合切片接入。工程回归不证明首次分数或长期收益已经改善。
diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py
index 6216d3dc46..c17c7afd62 100644
--- a/benchmark/runtime/harbor.py
+++ b/benchmark/runtime/harbor.py
@@ -73,6 +73,7 @@ def __init__(
         turn_envelope=False,
         native_goal_continue=False,
         task_objective=None,
+        research_loop=False,
         goal_id=_GOAL_ID,
         agent_id=_AGENT_ID,
         defer_codex_home=False,
@@ -112,6 +113,9 @@ def __init__(
         self.replan_after_todos = int(3 if replan_after_todos is None else replan_after_todos)
         if not 1 <= self.replan_after_todos <= 5:
             raise ValueError("replan_after_todos must be between 1 and 5")
+        from .research_loop import research_stride_strategy
+
+        stride_strategy = research_stride_strategy(research_loop, stride_strategy)
         if stride_strategy is not None and not self.execution.uses_loopx:
             raise ValueError("stride_strategy requires a LoopX execution mode")
         self.stride_strategy = (normalize_stride_strategy(stride_strategy)
@@ -122,6 +126,9 @@ def __init__(
             if not self.execution.uses_loopx:
                 raise ValueError("planning_horizon_goal requires a LoopX execution mode")
         self.planning_horizon_goal = planning_horizon_goal
+        if not isinstance(research_loop, bool) or (research_loop and self.execution.mode != "heartbeat"):
+            raise ValueError("research_loop requires heartbeat execution")
+        self.research_loop = research_loop
         self.checkpoint_request_path = None
         self._checkpoint_declaration = None
         if checkpoint_request is not None:
@@ -327,6 +334,12 @@ async def install(self, environment: BaseEnvironment) -> None:
                 raise RuntimeError("LoopX doctor returned invalid JSON") from exc
             if doctor_payload.get("ok") is not True:
                 raise RuntimeError(f"LoopX doctor failed: {doctor_payload}")
+            if self.research_loop:
+                wrapper = (f"#!/bin/sh\nexec env PYTHONPATH={shlex.quote(_SRC)} "
+                           f"{_PYTHON}/bin/python3 -m benchmark.runtime.research_probe \"$@\"\n")
+                await self.exec_as_agent(environment,
+                    command=f"printf %s {shlex.quote(wrapper)} > {_PROFILE}/bin/loopx-research-probe; "
+                            f"chmod 0755 {_PROFILE}/bin/loopx-research-probe", env=self._profile_env())
 
         receipt = {
             "loopx_commit": actual_commit,
@@ -335,6 +348,7 @@ async def install(self, environment: BaseEnvironment) -> None:
             "execution_mode": self.execution.mode,
             "iteration_context": self.execution.context,
             "task_entry": self.execution.task_entry,
+            **({"research_loop": True} if self.research_loop else {}),
             **self._turn_envelope_receipt(),
             **self._planning_receipt(),
             "home_scope": "trial",
@@ -494,8 +508,12 @@ async def _prepare_phase(
                 cwd=cwd,
             )
 
-        if self.execution.task_entry == "seeded-todo":
+        if self.execution.task_entry == "seeded-todo" or self.research_loop:
             await self._seed_phase(environment, cwd=cwd)
+        if self.research_loop:
+            from .research_loop import prepare_research_continuation
+
+            await prepare_research_continuation(self, environment, cwd=cwd)
 
         cadence = await self._loopx(
             environment, ["configure-goal", "--goal-id", self.goal_id], cwd=cwd,
@@ -521,6 +539,13 @@ async def _seed_phase(self, environment: BaseEnvironment, *, cwd: str) -> None:
             "implement and validate against the task's full requirements and acceptance "
             "criteria. Keep unmet requirements explicit when judging task completion."
         )
+        if self.research_loop:
+            text = (
+                f"[P0] Build and evaluate the strongest feasible initial candidate for {self._task_document}. "
+                "Check the actual interface, correctness and resource use before the first submission. "
+                "Preserve the tested candidate and released feedback. Completing this candidate stage "
+                "does not complete the separate original research work."
+            )
         if self._seeded_todo_id:
             listed = await self._loopx(environment, [
                 "todo", "list", "--goal-id", self.goal_id, "--role", "agent",
@@ -582,6 +607,7 @@ def _worker_env(self, *, cwd: str) -> dict[str, str]:
                 "LOOPX_SHARED_SKILLS": _SHARED_SKILLS,
                 "LOOPX_EXECUTION_MODE": self.execution.mode,
                 "LOOPX_TASK_ENTRY": self.execution.task_entry,
+                **({"LOOPX_RESEARCH_LOOP": "1"} if self.research_loop else {}),
                 **({"LOOPX_CHECKPOINT_REQUEST": self.checkpoint_request_path}
                    if self.checkpoint_request_path is not None else {}),
                 **({"LOOPX_TURN_ENVELOPE": "1"} if self.execution.turn_envelope else {}),
diff --git a/benchmark/runtime/planning.py b/benchmark/runtime/planning.py
index 93fa37163a..545a31ca68 100644
--- a/benchmark/runtime/planning.py
+++ b/benchmark/runtime/planning.py
@@ -36,7 +36,12 @@ def render_planning_prompt(packet: dict, env: dict[str, str]) -> str:
         value = float(env[key])
         if not math.isfinite(value) or value <= 0:
             raise ValueError(f"{key} must be finite and positive")
-    return ("$loopx\n\nBenchmark research planning guidance:\n" + _RESEARCH_PLANNING_GUIDANCE
+    research = ""
+    if env.get("LOOPX_RESEARCH_LOOP") == "1":
+        from .research_loop import research_execution_guidance
+
+        research = research_execution_guidance()
+    return ("$loopx\n\nBenchmark research planning guidance:\n" + _RESEARCH_PLANNING_GUIDANCE + research
             + "\n\nHost-supplied planning checkpoint:\n" + json.dumps(packet, ensure_ascii=False))
 
 
diff --git a/benchmark/runtime/research_loop.py b/benchmark/runtime/research_loop.py
new file mode 100644
index 0000000000..3f18edad20
--- /dev/null
+++ b/benchmark/runtime/research_loop.py
@@ -0,0 +1,73 @@
+"""Compose research scope with ordinary Todo/wait owners, not a second planner."""
+from __future__ import annotations
+
+
+def research_stride_strategy(enabled, strategy):
+    """One treatment definition shared by native CLI and programmatic entries."""
+    if not enabled:
+        return strategy
+    if strategy not in (None, "evidence_conditioned_v1"):
+        raise ValueError("research_loop requires evidence_conditioned_v1, not a disabled or different review strategy")
+    return "evidence_conditioned_v1"
+
+
+async def prepare_research_continuation(runtime, environment, *, cwd):
+    """A candidate stage cannot close a separately retained research outcome.
+
+    The deferred successor contains actual optimization work, not a recurring
+    instruction to replan. Its existing resume owner releases it when the
+    initial stage is done; budget, Stop and evidence-backed closeout stay native.
+    """
+    initial = runtime._seeded_todo_id
+    if not initial:
+        raise ValueError("research continuation requires its initial candidate stage")
+    listed = await runtime._loopx(environment, ["todo", "list", "--goal-id", runtime.goal_id,
+        "--role", "agent"], cwd=cwd)
+    matches = [todo for todo in listed["todos"] if todo.get("target_key") == "research-continuation"]
+    if len(matches) > 1:
+        raise ValueError("research continuation identity is ambiguous")
+    if matches:
+        current = matches[0]
+        if current.get("claimed_by") != runtime.agent_id:
+            raise ValueError("research continuation belongs to another Agent")
+        return current  # Never reopen completed work or clear a real wait on preparation.
+    research = await runtime._loopx(environment, [
+        "todo", "add", "--goal-id", runtime.goal_id, "--role", "agent",
+        "--claimed-by", runtime.agent_id, "--task-class", "advancement_task",
+        "--action-kind", "benchmark_task", "--target-key", "research-continuation",
+        "--status", "deferred", "--resume-when", "todo_done:" + initial,
+        "--text", f"[P1] Improve the original research objective in {runtime._task_document} through "
+        "complete, evidence-driven experiments. Preserve the best validated candidate. Use released "
+        "results and actual resource diagnostics to choose between a local repair and a different "
+        "mechanism; retain negative results. Candidate correctness and a successful submission settle "
+        "only that candidate stage. Close this research work only with evidence against the original "
+        "objective and remaining useful authorized work. Explicit user Stop, the original budget limit "
+        "and genuine blockers remain valid stops; do not manufacture work or run merely to fill time.",
+        "--execute"], cwd=cwd)
+    await runtime._loopx(environment, ["todo", "update", "--goal-id", runtime.goal_id,
+        "--agent-id", runtime.agent_id, "--todo-id", initial,
+        "--successor-todo-id", research["todo_id"], "--execute"], cwd=cwd)
+    await runtime._loopx(environment, ["refresh-state", "--goal-id", runtime.goal_id,
+        "--agent-id", runtime.agent_id, "--classification", "research_scope_initialized",
+        "--delivery-batch-scale", "single_surface", "--delivery-outcome", "surface_only",
+        "--vision-state", "active", "--vision-advancement-policy", "as_needed",
+        "--vision-summary", "Build a strong initial candidate, then execute evidence-driven improvements while retaining the best validated artifact.",
+        "--vision-acceptance", "Candidate validity settles its stage, not the original research objective. Respect user Stop, original budget and genuine blockers; justify research closeout from remaining useful authorized work.",
+        "--no-global-sync", "--suppress-external-sinks"], cwd=cwd)
+    return research
+
+
+def research_execution_guidance() -> str:
+    return (
+        '\nResearch-loop execution evidence:\nThe initial candidate and ongoing research are separate canonical Todos. '
+        'Do not close the research successor merely because the candidate passes tests. Preserve the original '
+        'objective and existing Stop/budget/blocked conditions. Before the first native submission, run a relevant '
+        'local candidate check and retain actual resource observations, including unknown measurements. Use '
+        '`loopx-research-probe --help` for the local probe wrapper: it pins explicit candidate files, measures '
+        'runtime and enforces the selected local limit. A solver can write search_iterations and stop_reason as JSON '
+        'to LOOPX_RESEARCH_DIAGNOSTICS; absent values remain unknown, and solver reports are not independently '
+        'verified. This is not another evaluation or a requirement to use every available second. Attach the emitted '
+        'execution_observation in the normal Explore result writeback. Preserve its input revision and evidence '
+        'refs. Record changed method ids and the reason when actual implementation differs from the proposal. '
+        'Resolve this evidence at the next replan.\n'
+    )
diff --git a/benchmark/runtime/research_probe.py b/benchmark/runtime/research_probe.py
new file mode 100644
index 0000000000..a8a913e0dc
--- /dev/null
+++ b/benchmark/runtime/research_probe.py
@@ -0,0 +1,160 @@
+"""Run a caller-selected local probe and emit an Explore execution attachment.
+
+No evaluator access or submission is performed. Candidate file pins cover only
+the explicitly selected files at the probe boundaries, not transitive imports.
+The optional solver diagnostic is reported separately from host measurements.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import math
+import os
+from pathlib import Path
+import signal
+import subprocess
+import time
+
+
+def candidate_snapshot(paths: list[Path], workspace: Path) -> tuple[str, list[dict[str, str]]]:
+    files = []
+    for path in paths:
+        resolved = path.resolve(strict=True)
+        if not resolved.is_relative_to(workspace) or not resolved.is_file():
+            raise ValueError("candidate files must be existing files in the workspace")
+        with resolved.open("rb") as stream:
+            digest = hashlib.file_digest(stream, "sha256").hexdigest()
+        files.append({"path": resolved.relative_to(workspace).as_posix(), "sha256": digest})
+    if not 1 <= len(files) <= 32 or len({row["path"] for row in files}) != len(files):
+        raise ValueError("select one to 32 distinct candidate files")
+    files.sort(key=lambda row: row["path"])
+    encoded = json.dumps(files, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(encoded.encode()).hexdigest(), files
+
+
+def run_probe(*, command: list[str], candidate_files: list[Path], output: Path,
+              input_revision: str, planned_method_id: str, executed_method_id: str,
+              evidence_ref: str, limit_seconds: float, change_reason: str | None = None,
+              diagnostics: Path | None = None) -> dict:
+    from loopx.capabilities.explore.execution_observation import normalize_execution_observation
+
+    if not command or not math.isfinite(limit_seconds) or limit_seconds <= 0:
+        raise ValueError("a command and finite positive local probe limit are required")
+    workspace = Path.cwd().resolve()
+    before, files = candidate_snapshot(candidate_files, workspace)
+    observation = dict(schema_version="explore_execution_observation_v0",
+        candidate_sha256=before, input_revision=input_revision, planned_method_id=planned_method_id,
+        executed_method_id=executed_method_id, change_reason=change_reason,
+        elapsed_seconds=None, limit_seconds=limit_seconds, search_iterations=None,
+        stop_reason=None, exit_code=None, evidence_refs=[evidence_ref, "source:" + before])
+    normalize_execution_observation(observation)  # Reject bad attribution before execution.
+    if diagnostics is not None and diagnostics.exists():
+        raise FileExistsError("use a fresh diagnostic path; previous observations cannot describe this probe")
+    # Reserve before any process effect. Existing evidence is never overwritten.
+    with output.open("x", encoding="utf-8") as record:
+        with output.with_suffix(output.suffix + ".candidate.json").open("x", encoding="utf-8") as source:
+            source.write(json.dumps({"candidate_sha256": before, "files": files}, sort_keys=True) + "\n")
+        env = dict(os.environ)
+        if diagnostics is not None:
+            env["LOOPX_RESEARCH_DIAGNOSTICS"] = str(diagnostics.resolve())
+        started = time.monotonic()
+        timed_out = False
+        with output.with_suffix(output.suffix + ".stdout").open("xb") as stdout, \
+             output.with_suffix(output.suffix + ".stderr").open("xb") as stderr:
+            process = subprocess.Popen(command, cwd=workspace, env=env, stdout=stdout, stderr=stderr,
+                                       start_new_session=True)
+            try:
+                process.wait(timeout=limit_seconds)
+            except subprocess.TimeoutExpired:
+                timed_out = True
+            finally:
+                # Stop the owned group, including descendants that outlive a
+                # successful leader; they must not continue changing a candidate.
+                try:
+                    os.killpg(process.pid, signal.SIGKILL)
+                except ProcessLookupError:
+                    pass
+                process.wait()
+        observation["elapsed_seconds"] = time.monotonic() - started
+        observation["exit_code"] = process.returncode
+        observation["stop_reason"] = "host_timeout" if timed_out else f"process_exit:{process.returncode}"
+        after, _ = candidate_snapshot(candidate_files, workspace)
+        if after != before:
+            record.write(json.dumps({"ok": False, "reason": "candidate_changed_during_probe",
+                                     "before_sha256": before, "after_sha256": after}))
+            raise ValueError("candidate changed during the probe; preserve this attempt and rerun a fixed candidate")
+        # Host measurements stand independently of the optional solver report.
+        # Validate them first, so a malformed/unavailable report cannot erase a
+        # completed execution or change its real exit/timeout status.
+        result = normalize_execution_observation(observation)
+        try:
+            if diagnostics is not None:
+                try:
+                    if diagnostics.exists():
+                        if not diagnostics.is_file():
+                            raise ValueError("solver diagnostics must be a regular file")
+                        with diagnostics.open("rb") as stream:
+                            raw = stream.read(4097)
+                        if len(raw) > 4096:
+                            raise ValueError("solver diagnostics exceed the bounded observation size")
+                        reported = json.loads(raw)
+                        if not isinstance(reported, dict) or set(reported) - {"search_iterations", "stop_reason"}:
+                            raise ValueError("solver diagnostics accept only search_iterations and stop_reason")
+                        proposed = dict(result, search_iterations=reported.get("search_iterations"))
+                        reason = reported.get("stop_reason")
+                        if reason is not None:
+                            if not isinstance(reason, str):
+                                raise ValueError("solver stop_reason must be text or null")
+                            proposed["stop_reason"] += "; solver_report:" + reason
+                        checked = normalize_execution_observation(proposed)
+                        if timed_out:
+                            checked["stop_reason"] = "host_timeout"
+                        result = checked
+                except Exception as error:
+                    # Confined to optional diagnostics, never candidate checks
+                    # or process execution. Retain a bounded local explanation
+                    # without copying raw solver text or exception paths.
+                    detail = dict(source="solver_diagnostics", error_kind=type(error).__name__,
+                        candidate_sha256=before, probe_evidence_ref=evidence_ref,
+                        message="Optional solver diagnostics are invalid or unavailable; solver measurements remain unknown.")
+                    encoded = json.dumps(detail, sort_keys=True, separators=(",", ":")).encode() + b"\n"
+                    with output.with_suffix(output.suffix + ".diagnostics-error.json").open("xb") as error_record:
+                        error_record.write(encoded)
+                    result["evidence_refs"].append("diagnostics-error:" + hashlib.sha256(encoded).hexdigest())
+                    result = normalize_execution_observation(result)
+        finally:
+            # Even failure to retain the diagnostic explanation leaves the
+            # independently valid host observation available to its caller.
+            record.write(json.dumps(result, ensure_ascii=False, allow_nan=False) + "\n")
+            record.flush()
+            os.fsync(record.fileno())
+    return result
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--candidate-file", type=Path, action="append", required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--input-revision", required=True)
+    parser.add_argument("--planned-method-id", required=True)
+    parser.add_argument("--executed-method-id", required=True)
+    parser.add_argument("--change-reason")
+    parser.add_argument("--evidence-ref", required=True)
+    parser.add_argument("--limit-seconds", type=float, required=True)
+    parser.add_argument("--diagnostics", type=Path)
+    parser.add_argument("command", nargs=argparse.REMAINDER)
+    args = vars(parser.parse_args())
+    args["candidate_files"] = args.pop("candidate_file")
+    if args["command"][:1] == ["--"]:
+        args["command"] = args["command"][1:]
+    result = run_probe(**args)
+    print(json.dumps(result, allow_nan=False))
+    if result["stop_reason"] == "host_timeout":
+        return 124
+    code = result["exit_code"]
+    return 128 - code if code < 0 else code
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py
index 1ddeef2bc3..ad62f7f07a 100644
--- a/benchmark/runtime/worker.py
+++ b/benchmark/runtime/worker.py
@@ -85,9 +85,14 @@ def heartbeat_body(env: dict[str, str], turn_id: str, *, native_goal: bool,
         raise RuntimeError("formal LoopX heartbeat renderer is not ready")
     if not native_goal and payload.get("turn_instance_id") != turn_id:
         raise RuntimeError("heartbeat Turn identity mismatch")
-    return payload["task_body"].replace(
+    body = payload["task_body"].replace(
         "$HOME/.codex/loopx/registry.global.json", env["LOOPX_REGISTRY"]
     )
+    if env.get("LOOPX_RESEARCH_LOOP") == "1":
+        from .research_loop import research_execution_guidance
+
+        body += research_execution_guidance()
+    return body
 
 
 @contextmanager
diff --git a/docs/assets/research-execution/desktop.png b/docs/assets/research-execution/desktop.png
new file mode 100644
index 0000000000..a5e8c95e77
Binary files /dev/null and b/docs/assets/research-execution/desktop.png differ
diff --git a/docs/assets/research-execution/narrow.png b/docs/assets/research-execution/narrow.png
new file mode 100644
index 0000000000..29ad0c3a55
Binary files /dev/null and b/docs/assets/research-execution/narrow.png differ
diff --git a/docs/reference/protocols/goal-vision-replan-contract-v0.md b/docs/reference/protocols/goal-vision-replan-contract-v0.md
index be25236830..9fab3977e8 100644
--- a/docs/reference/protocols/goal-vision-replan-contract-v0.md
+++ b/docs/reference/protocols/goal-vision-replan-contract-v0.md
@@ -327,6 +327,21 @@ plan invalidates its unexecuted proposal, not the unresolved source. This is a
 read model, not a pending-duty ledger, new write gate or forced continuation.
 The Agent still checks the provider's exact candidate/input/result revision.
 
+The relation uses the existing delivery outcome owner: only `outcome_gap`,
+`outcome_progress` and `primary_goal_outcome` records create pending work results
+or satisfy subsequent followthrough. `surface_only`, missing and unsupported
+historical outcomes remain readable in ordinary history but neither consume a
+cited result nor create a new unresolved result. An optional progress attachment
+does not override that classification. Negative results and concrete blockers
+remain eligible material evidence; this does not certify execution or gain,
+rewrite source history, or alter settlement, waiting or continuation authority.
+
+结果接续复用既有 delivery outcome 分类：只有 `outcome_gap`、`outcome_progress`
+和 `primary_goal_outcome` 才形成待复盘工作结果或后续接续关系。纯整理
+`surface_only`、缺失或未知历史分类仍可从普通历史读取，但不会把旧结果标为已接续，
+也不会产生新的待复盘结果；可选 progress 附件不能覆盖这一分类。有效负结果和明确
+阻塞仍保留。这只是来源关系，不认证实际执行或收益，不改写历史、结算或继续运行权限。
+
 启用 `evidence_conditioned_v1` 且历史中已有被接受、绑定证据的 Vision 路线时，replan context 返回上次决定与之后的
 进度引用，按合格来源的顺序关联，而非根据时间戳或文案猜测因果。完整本地索引提供
 追加位置；已知最新在前的紧凑载体与 `newest_first_runs` 入口转换位置，保留普通
diff --git a/examples/personal-workspace-browser/research-history.mjs b/examples/personal-workspace-browser/research-history.mjs
index db7283ec41..ee8fc33fee 100644
--- a/examples/personal-workspace-browser/research-history.mjs
+++ b/examples/personal-workspace-browser/research-history.mjs
@@ -235,7 +235,13 @@ events=[build_explore_node_event(goal_id=goal,node_id=f'hit-{index}',title=f'Syn
     summary='Pointer regression only',parent_id='hit-0' if index else None,
     recorded_at='2026-01-01T00:00:00Z' if index<9 else '2026-01-01T18:00:00Z') for index in range(10)]
 events.append(build_explore_finding_event(goal_id=goal,finding_id='negative',node_id='hit-6',
-    title='Synthetic negative',summary='Negative pointer regression',status='refuted',recorded_at='2026-01-01T00:00:00Z'))
+    title='Synthetic negative',summary='Negative pointer regression',status='refuted',recorded_at='2026-01-01T00:00:00Z',
+    evidence_refs=['probe:synthetic'], execution_observation={
+        'schema_version':'explore_execution_observation_v0','candidate_sha256':'a'*64,
+        'input_revision':'synthetic-v1','planned_method_id':'scheduler','executed_method_id':'mask-selection',
+        'change_reason':'The control isolated a different first intervention.',
+        'elapsed_seconds':0.2,'limit_seconds':30,'search_iterations':None,'exit_code':0,
+        'stop_reason':'process_exit:0','evidence_refs':['probe:synthetic']}))
 Path(sys.argv[1]).write_text(json.dumps({'goal_id':goal,'display_name':'Synthetic pointer regression','events':events}))
 `,fixture],{cwd:repoRoot});
       collisionOwner=await startHistoryOwner(fixture);
@@ -294,6 +300,11 @@ Path(sys.argv[1]).write_text(json.dumps({'goal_id':goal,'display_name':'Syntheti
         await page.mouse.click(box.x + box.width / 2, box.y + box.height / 2);
         const detail = history.locator('.research-history-detail');
         await detail.waitFor();
+        await detail.getByText('实际执行与资源观测', {exact:true}).click();
+        assert(await detail.getByText('scheduler → mask-selection', {exact:true}).isVisible());
+        assert(await detail.getByText('未测量', {exact:true}).isVisible(), 'Unknown iterations must not become zero');
+        assert(await detail.getByText('0.2 / 30', {exact:true}).isVisible());
+        await page.screenshot({path:resolve(outputDir, `research-execution-${viewport.width}.png`), animations:'disabled'});
         assert.deepEqual(selectedNodes(),ids);
         assert(ids.includes('hit-6'));
         await detail.getByRole('heading',{name:'Synthetic hit target 6',exact:true}).waitFor();
diff --git a/loopx/capabilities/explore/README.md b/loopx/capabilities/explore/README.md
index d30ea8c99b..106b7c9da5 100644
--- a/loopx/capabilities/explore/README.md
+++ b/loopx/capabilities/explore/README.md
@@ -2,6 +2,16 @@
 
 Status: supported optional capability; default-off harness execution contract.
 
+Ordinary result writeback can include an optional `execution_observation`
+(`explore_execution_observation_v0`): exact candidate SHA-256 and input revision,
+planned/executed method ids and change reason, measured elapsed/limit seconds,
+search iterations, stop reason and opaque evidence refs. Unknown measurements
+are explicit nulls. Its refs and input revision must belong to the enclosing
+finding. Released `result_ref` requires a matching `result_candidate_sha256`.
+This records producer evidence, not verified execution or utility. Findings,
+replan context and the existing research-history detail retain it; omission
+preserves old events. See the [local probe producer](../../../benchmark/runtime/RESEARCH_LOOP.md).
+
 ## At a Glance
 
 Explore Harness is a supported, default-off optional capability for
diff --git a/loopx/capabilities/explore/README.zh-CN.md b/loopx/capabilities/explore/README.zh-CN.md
index fc72ece36a..2d643c1110 100644
--- a/loopx/capabilities/explore/README.zh-CN.md
+++ b/loopx/capabilities/explore/README.zh-CN.md
@@ -2,6 +2,13 @@
 
 状态：受支持的 optional capability；harness 执行契约默认关闭。
 
+普通结果写回可附带可选 `execution_observation`（`explore_execution_observation_v0`）：
+候选 SHA-256、输入版本、计划/实际方法标识、改向原因、实测耗时/上限、搜索次数、
+结束原因与证据引用。未测量项显式为 null；版本和引用必须与所属 finding 一致。
+已释放结果的 `result_ref` 必须有匹配的 `result_candidate_sha256`。该记录是来源证据，
+不认证执行真实性或收益；finding、replan 和既有研究历程详情保留相同信息，缺省仍兼容
+旧事件。参见[本地探针生产路径](../../../benchmark/runtime/RESEARCH_LOOP.md)。
+
 ## 概览
 
 LoopX Explore 是面向长程探索目标（软件研究、安全攻击面测绘、领域调研）的受支持、默认关闭的可选能力。它把“到处看看”变成有界、可观测、有门控的流程，包含三根支柱：
diff --git a/loopx/capabilities/explore/execution_observation.py b/loopx/capabilities/explore/execution_observation.py
new file mode 100644
index 0000000000..bf5364baae
--- /dev/null
+++ b/loopx/capabilities/explore/execution_observation.py
@@ -0,0 +1,21 @@
+"""Transport/cache for the Explore-owned typed execution observation codec."""
+from __future__ import annotations
+
+from functools import lru_cache
+import json
+
+from ...control_plane.effect_runtime import effect_runtime_result
+from ...control_plane.runtime.public_safety import validate_public_safe_value
+
+
+@lru_cache(maxsize=512)
+def _normalize(encoded: str) -> str:
+    result = effect_runtime_result("explore.execution.normalize", {"observation": json.loads(encoded)})
+    return json.dumps(result, sort_keys=True, allow_nan=False)
+
+
+def normalize_execution_observation(value):
+    encoded = json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False)
+    validate_public_safe_value(value, path="execution_observation")
+    # Do not let readers mutate the cached canonical source.
+    return json.loads(_normalize(encoded))
diff --git a/loopx/capabilities/explore/result_log.py b/loopx/capabilities/explore/result_log.py
index 903026f458..d82c41e440 100644
--- a/loopx/capabilities/explore/result_log.py
+++ b/loopx/capabilities/explore/result_log.py
@@ -349,6 +349,7 @@ def build_explore_finding_event(
     tags: Sequence[str] | None = None,
     supersedes: str | None = None,
     recorded_at: str | None = None,
+    execution_observation: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
     safe_goal_id = _safe_goal_id(goal_id)
     safe_title = _compact_text(title, limit=TITLE_LIMIT, field="title")
@@ -379,6 +380,13 @@ def build_explore_finding_event(
     safe_confidence = _safe_confidence(confidence)
     if safe_confidence is not None:
         event["confidence"] = safe_confidence
+    if execution_observation is not None:
+        from .execution_observation import normalize_execution_observation
+
+        observed = normalize_execution_observation(execution_observation)
+        if any(ref not in event.get("evidence_refs", []) for ref in observed["evidence_refs"]):
+            raise ValueError("execution evidence must be included in the finding's evidence_refs")
+        event["execution_observation"] = observed
     event["event_id"] = _event_id(event)
     return event
 
@@ -497,6 +505,7 @@ def validate_explore_result_event(
             evidence_refs=payload.get("evidence_refs"),
             tags=payload.get("tags"),
             supersedes=payload.get("supersedes"),
+            execution_observation=payload.get("execution_observation"),
         )
     if summary:
         rebuilt["summary"] = summary
@@ -719,7 +728,7 @@ def explore_history_rows(events, *, goal_id=None):
         "schema_version", "event_id", "goal_id", "event_kind", "result_id",
         "recorded_at", "title", "summary", "status", "node_kind",
         "blocked_reason", "parent_id", "node_id", "from_node", "to_node",
-        "edge_type", "agent_id", "run_id", "evidence_refs", "supersedes", "confidence",
+        "edge_type", "agent_id", "run_id", "evidence_refs", "supersedes", "confidence", "execution_observation",
     )
     return [{key: event[key] for key in fields if key in event} for event in events]
 
@@ -820,6 +829,7 @@ def _materialized_parent_edges(
 
 def _finding_view(event: Mapping[str, Any]) -> dict[str, Any]:
     return {
+        **({"execution_observation": event["execution_observation"]} if "execution_observation" in event else {}),
         "finding_id": str(event.get("result_id") or ""),
         "finding": str(event.get("title") or ""),
         "summary": str(event.get("summary") or ""),
diff --git a/loopx/capabilities/explore/result_writeback.py b/loopx/capabilities/explore/result_writeback.py
index b8719e3420..8497e13fdb 100644
--- a/loopx/capabilities/explore/result_writeback.py
+++ b/loopx/capabilities/explore/result_writeback.py
@@ -133,6 +133,7 @@ def _events(attachment, *, goal_id, agent_id, source_id, recorded_at, existing_n
         ),
         evidence_refs=attachment["evidence_refs"],
         tags=["writeback-result"],
+        execution_observation=attachment.get("execution_observation"),
         recorded_at=recorded_at,
     )
     relations = [build_explore_edge_event(
diff --git a/loopx/control_plane/capabilities/explore_execution_observation.ts b/loopx/control_plane/capabilities/explore_execution_observation.ts
new file mode 100644
index 0000000000..c66d4e0956
--- /dev/null
+++ b/loopx/control_plane/capabilities/explore_execution_observation.ts
@@ -0,0 +1,78 @@
+/** Candidate execution facts carried by an existing Explore finding.
+ * The producer owns measurement; this codec validates attribution, not utility.
+ */
+import type {JsonObject} from "../runtime_decode.ts";
+import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
+import {requireJsonObject} from "../runtime_decode.ts";
+import {BARE_SHA256_PATTERN} from "../content_digest.ts";
+
+export function normalizeExploreExecutionObservation(params: JsonObject): JsonObject {
+  const row = requireJsonObject(params.observation, "execution observation");
+  const required = ["schema_version", "candidate_sha256", "input_revision", "planned_method_id",
+    "executed_method_id", "change_reason", "elapsed_seconds", "limit_seconds",
+    "search_iterations", "stop_reason", "evidence_refs"];
+  if (row.schema_version !== "explore_execution_observation_v0"
+    || required.some(key => !Object.hasOwn(row, key))
+    || Object.keys(row).some(key => ![...required, "exit_code", "result_ref", "result_candidate_sha256"].includes(key))) {
+    throw new EffectRuntimeRequestError("execution observation fields are missing or unsupported");
+  }
+  const text = (value: unknown, field: string, max: number): string => {
+    if (typeof value !== "string" || !value.trim() || value.length > max || /[\u0000-\u001f]/.test(value)) {
+      throw new EffectRuntimeRequestError(`${field} requires bounded single-line text`);
+    }
+    return value.trim();
+  };
+  const id = (value: unknown, field: string): string => {
+    const result = text(value, field, 128);
+    if (!/^[A-Za-z0-9][A-Za-z0-9_.:-]*$/.test(result)) {
+      throw new EffectRuntimeRequestError(`${field} requires an opaque identifier`);
+    }
+    return result;
+  };
+  const sha = (value: unknown): string => {
+    if (typeof value !== "string" || !BARE_SHA256_PATTERN.test(value)) {
+      throw new EffectRuntimeRequestError("execution candidate requires its exact SHA-256");
+    }
+    return value;
+  };
+  const measured = (value: unknown, field: string, integer = false, positive = false): number | null => {
+    if (value === null) return null;
+    if (typeof value !== "number" || !Number.isFinite(value) || value < 0 || (positive && value === 0)
+      || (integer && !Number.isSafeInteger(value))) {
+      throw new EffectRuntimeRequestError(`${field} requires a nonnegative measurement or explicit null`);
+    }
+    return value;
+  };
+  const planned = id(row.planned_method_id, "planned_method_id");
+  const executed = id(row.executed_method_id, "executed_method_id");
+  const change = row.change_reason === null ? null : text(row.change_reason, "change_reason", 240);
+  if (planned !== executed && change === null) {
+    throw new EffectRuntimeRequestError("a different executed method requires its change_reason");
+  }
+  if (!Array.isArray(row.evidence_refs) || !row.evidence_refs.length || row.evidence_refs.length > 8) {
+    throw new EffectRuntimeRequestError("execution observation requires one to eight evidence references");
+  }
+  const result: JsonObject = {schema_version: row.schema_version,
+    candidate_sha256: sha(row.candidate_sha256), input_revision: text(row.input_revision, "input_revision", 160),
+    planned_method_id: planned, executed_method_id: executed, change_reason: change,
+    elapsed_seconds: measured(row.elapsed_seconds, "elapsed_seconds"),
+    limit_seconds: measured(row.limit_seconds, "limit_seconds", false, true),
+    search_iterations: measured(row.search_iterations, "search_iterations", true),
+    stop_reason: row.stop_reason === null ? null : text(row.stop_reason, "stop_reason", 160),
+    evidence_refs: [...new Set(row.evidence_refs.map(ref => id(ref, "execution evidence_ref")))].sort()};
+  if (Object.hasOwn(row, "exit_code")) {
+    if (row.exit_code !== null && (typeof row.exit_code !== "number" || !Number.isSafeInteger(row.exit_code))) {
+      throw new EffectRuntimeRequestError("exit_code requires an integer process status or null");
+    }
+    result.exit_code = row.exit_code;
+  }
+  if (Object.hasOwn(row, "result_ref") || Object.hasOwn(row, "result_candidate_sha256")) {
+    result.result_ref = id(row.result_ref, "result_ref");
+    result.result_candidate_sha256 = sha(row.result_candidate_sha256);
+    if (result.result_candidate_sha256 !== result.candidate_sha256
+      || !(result.evidence_refs as string[]).includes(String(result.result_ref))) {
+      throw new EffectRuntimeRequestError("result evidence must name this executed candidate and an included evidence reference");
+    }
+  }
+  return result;
+}
diff --git a/loopx/control_plane/capabilities/explore_result_writeback.ts b/loopx/control_plane/capabilities/explore_result_writeback.ts
index 6b2b8c8b1e..b189f886f1 100644
--- a/loopx/control_plane/capabilities/explore_result_writeback.ts
+++ b/loopx/control_plane/capabilities/explore_result_writeback.ts
@@ -4,6 +4,7 @@ import type {JsonObject} from "../effect_program.ts";
 import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
 import {requireJsonObject, requireNonEmptyString, requireStringArray, requireStringLiteral} from "../runtime_decode.ts";
 import {normalizeGoalPathDelta} from "../goals/vision_checkpoint.ts";
+import {normalizeExploreExecutionObservation} from "./explore_execution_observation.ts";
 
 export const EXPLORE_RESULT_ATTACHMENT_SCHEMA = "explore_result_attachment_v0";
 export const EXPLORE_PATH_DELTA_ATTACHMENT_SCHEMA = "explore_result_from_path_delta_v0";
@@ -58,6 +59,7 @@ export function exploreResultWritebackAffordance(): JsonObject {
       "Capture is optional, not a settlement obligation.",
       "A stopped route does not imply refuted status, and a score alone does not prove refutation.",
       "Keep raw logs local.",
+      "When an experiment has execution evidence, attach execution_observation: preserve its exact candidate and input version, planned versus executed method, change reason and measured resources. Missing measurements are null, not zero. A finding source digest is not an executable digest. Result evidence must name the candidate it actually evaluated, including delayed or regressing results; it does not certify current workspace quality or Goal completion.",
       "Observations allow 320 characters, interpretations 1200; the complete scoped summary is preserved within 2000 characters.",
     ].join(" "),
     // Blank evidence fields deliberately fail validation until the caller
@@ -72,7 +74,7 @@ export function exploreResultWritebackAffordance(): JsonObject {
 }
 
 const FIELDS = ["schema_version", "node_id", "question", "applicability", "input_revision",
-  "observation", "interpretation", "status", "evidence_refs", "question_update", "relations"];
+  "observation", "interpretation", "status", "evidence_refs", "question_update", "relations", "execution_observation"];
 // Reuse the persisted Explore node/edge vocabulary, distinct from finding status.
 const QUESTION_STATUSES = ["open", "exploring", "blocked", "resolved", "dead_end"] as const;
 const RELATION_TYPES = ["subtopic_of", "depends_on", "answers", "supports", "refutes", "leads_to"] as const;
@@ -153,6 +155,12 @@ function normalizeAttachment(value: unknown, visionPacket?: unknown, linkedScope
     }
     return ref;
   });
+  const execution = Object.hasOwn(row, "execution_observation")
+    ? normalizeExploreExecutionObservation({observation: row.execution_observation}) : null;
+  if (execution && (execution.input_revision !== text(row.input_revision, "input_revision", 160)
+    || (execution.evidence_refs as string[]).some(ref => !refs.includes(ref)))) {
+    throw new EffectRuntimeRequestError("execution observation must retain this finding's input revision and evidence references");
+  }
   // Status reuses the existing finding vocabulary. The caller supplies the
   // interpretation; a failed prerequisite belongs in a tentative observation.
   return {schema_version: EXPLORE_RESULT_ATTACHMENT_SCHEMA, node_id: node,
@@ -161,11 +169,12 @@ function normalizeAttachment(value: unknown, visionPacket?: unknown, linkedScope
     observation: text(row.observation, "observation", 320),
     interpretation: text(row.interpretation, "interpretation", 1200),
     status: requireStringLiteral(row.status, ["tentative", "confirmed", "refuted"], "finding status"),
-    evidence_refs: [...new Set(refs)], ...evolutionFields(row, node)};
+    evidence_refs: [...new Set(refs)], ...evolutionFields(row, node),
+    ...(execution ? {execution_observation: execution} : {})};
 }
 /** Explicit reference, not inferred evidence. Preserve every route item verbatim. */
 function attachmentFromPathDelta(row: JsonObject, visionPacket: unknown, linkedScope: unknown): JsonObject {
-  const fields = ["schema_version", "node_id", "question", "applicability", "input_revision", "status", "evidence_refs", "question_update", "relations"];
+  const fields = ["schema_version", "node_id", "question", "applicability", "input_revision", "status", "evidence_refs", "question_update", "relations", "execution_observation"];
   if (Object.keys(row).some(key => !fields.includes(key))) {
     throw new EffectRuntimeRequestError("Explore path_delta attachment contains unknown fields");
   }
diff --git a/loopx/control_plane/capabilities/explore_turn_context.ts b/loopx/control_plane/capabilities/explore_turn_context.ts
index fcc234e02b..498b6134c4 100644
--- a/loopx/control_plane/capabilities/explore_turn_context.ts
+++ b/loopx/control_plane/capabilities/explore_turn_context.ts
@@ -178,6 +178,7 @@ export function projectExploreTurnContext(params: JsonObject): JsonObject {
       writeback_results: visibleResults
         .map(row => ({...compact(row, ["finding_id", "node_id", "finding", "status", "evidence_refs"]),
           summary: String(row.summary ?? "").slice(0, EXPLORE_WRITEBACK_SUMMARY_LIMIT),
+          ...(row.execution_observation ? {execution_observation: row.execution_observation} : {}),
           ...currentQuestion(row),
           ...findingDigest(row, findingSources)})),
       result_page: {
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index aab011ea35..c6bd792359 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -802,6 +802,7 @@ export function createEffectRuntimeHandlers(
     ["explore.configuration.resolve", lazyHandler(() => import("./capabilities/explore_configuration.ts"), ({resolveExploreConfiguration}) => resolveExploreConfiguration)],
     ["explore.configuration.plan", lazyHandler(() => import("./capabilities/explore_configuration.ts"), ({planExploreConfiguration}) => planExploreConfiguration)],
     ["explore.result.normalize", lazyHandler(() => import("./capabilities/explore_result_writeback.ts"), ({normalizeExploreResultAttachment}) => normalizeExploreResultAttachment)],
+    ["explore.execution.normalize", lazyHandler(() => import("./capabilities/explore_execution_observation.ts"), ({normalizeExploreExecutionObservation}) => normalizeExploreExecutionObservation)],
     ["explore.result.evolution", lazyHandler(() => import("./capabilities/explore_result_writeback.ts"), ({planExploreQuestionEvolution}) => planExploreQuestionEvolution)],
     ["explore.result.intent", lazyHandler(() => import("./capabilities/explore_result_writeback.ts"), ({produceExploreResultIntent}) => produceExploreResultIntent)],
     ["explore.turn_context", lazyHandler(() => import("./capabilities/explore_turn_context.ts"), ({projectExploreTurnContext}) => projectExploreTurnContext)],
diff --git a/loopx/control_plane/presentation/explore_history.ts b/loopx/control_plane/presentation/explore_history.ts
index d161650c9b..bde2c9f296 100644
--- a/loopx/control_plane/presentation/explore_history.ts
+++ b/loopx/control_plane/presentation/explore_history.ts
@@ -2,12 +2,15 @@
  * Event kinds/statuses/relationships remain owned by Explore result_log.py.
  * Local display grouping must not turn parentage or time into supporting evidence.
  */
+import {normalizeExploreExecutionObservation} from "../capabilities/explore_execution_observation.ts";
+import type {JsonObject} from "../runtime_decode.ts";
 export type ExploreHistoryEvent = {
   schema_version: string; event_id: string; goal_id: string; event_kind: string;
   result_id: string; recorded_at: string; title: string; summary: string;
   status: string; node_kind: string; blocked_reason: string; parent_id: string;
   node_id: string; from_node: string; to_node: string; edge_type: string;
   agent_id: string; run_id: string; evidence_refs: string[]; supersedes: string; confidence: number | null;
+  execution_observation?: JsonObject;
 };
 export type ExploreHistoryPage = {goal_id: string; items: ExploreHistoryEvent[]; total: number; next_cursor: string | null};
 export type ExploreHistoryNode = {id: string; latest: ExploreHistoryEvent; updates: ExploreHistoryEvent[]; findings: ExploreHistoryEvent[]; revisions: ExploreHistoryEvent[]};
@@ -58,6 +61,9 @@ export function parseExploreHistoryPage(input: unknown, goalId: string): Explore
       ...Object.fromEntries(fields.map(key => [key, raw[key] ?? ""])),
       evidence_refs: [...(raw.evidence_refs as string[] | undefined ?? [])],
       confidence: typeof raw.confidence === "number" ? raw.confidence : null,
+      ...(raw.execution_observation === undefined ? {} : {
+        execution_observation: normalizeExploreExecutionObservation({observation: raw.execution_observation}),
+      }),
     } as ExploreHistoryEvent;
   });
   return {goal_id: goalId, items, total: page.total as number, next_cursor: page.next_cursor as string | null};
diff --git a/loopx/control_plane/work_items/replan_context.ts b/loopx/control_plane/work_items/replan_context.ts
index f909d478ad..3627cd7cb0 100644
--- a/loopx/control_plane/work_items/replan_context.ts
+++ b/loopx/control_plane/work_items/replan_context.ts
@@ -9,6 +9,7 @@ import {FRESH_PATH_DISPOSITIONS, requiredSemanticOutcomes} from "./replan_semant
 import {readReplanSnapshot} from "./replan_history_snapshot.ts";
 import {normalizeGoalPathDelta} from "../goals/vision_checkpoint.ts";
 import {normalizeRecordedStrideDecision, normalizeStrideStrategy, projectStrideExploreBasis} from "./stride_policy.ts";
+import {MATERIAL_DELIVERY_OUTCOMES} from "./delivery_outcome.ts";
 
 const CONTEXT_SCHEMA = "replan_context_v0";
 // Replan is an infrequent decision, not the ordinary guard's display window.
@@ -229,6 +230,12 @@ function workResultUse(scoped: JsonObject[]): JsonObject | null {
       }
       continue;
     }
+    // Reuse the delivery owner's recorded classification. A status/preparation
+    // writeback is still readable evidence, but cannot consume a cited result
+    // or create a new pending work result. Unknown legacy outcomes stay unknown.
+    // Material gaps remain eligible: this relation certifies neither successful
+    // execution nor scientific gain, and must preserve useful negative evidence.
+    if (!MATERIAL_DELIVERY_OUTCOMES.some(outcome => outcome === row.delivery_outcome)) continue;
     const todo = String(observationTodo(row));
     if (!origin.has(todo)) continue;
     const ref = "replan-evidence-" + digest(readableEvidenceRow(row));
@@ -250,7 +257,7 @@ function workResultUse(scoped: JsonObject[]): JsonObject | null {
     unresolved: visible, recorded_followthrough_count: followed.length,
     recorded_followthrough: followed.slice(-6).reverse(),
     omitted_followthrough_count: Math.max(0, followed.length - 6),
-    guidance: "Resolve unresolved evidence even when a newer review has occurred. A cited result stays unresolved until later recorded work is linked through that decision's declared Todo. " +
+    guidance: "Resolve unresolved evidence even when a newer review has occurred. A cited result stays unresolved until a later material delivery is linked through that decision's declared Todo; surface-only and unknown outcomes do not consume it. " +
       "Recorded followthrough is an attributable history relation, not verified execution, applicability to a current candidate, or Goal gain. " +
       "Check the provider's exact candidate/input/result revision before adopting an outcome; retain valid negative and inconclusive evidence. Existing authority, stopping and settlement rules remain unchanged."};
 }
diff --git a/loopx/control_plane/work_items/stride_policy.ts b/loopx/control_plane/work_items/stride_policy.ts
index 5176b76e12..dcb588e0ad 100644
--- a/loopx/control_plane/work_items/stride_policy.ts
+++ b/loopx/control_plane/work_items/stride_policy.ts
@@ -239,6 +239,7 @@ export function projectStrideExploreBasis(value: unknown, goal: string, agent: s
   for (const row of [...findings, ...audits, ...recent]) {
     const id = requireNonEmptyString(row.finding_id, "Explore finding id");
     if (!byId.has(id)) byId.set(id, {finding_id: id, node_id: row.node_id, status: row.status,
+      ...(row.execution_observation ? {execution_observation: row.execution_observation} : {}),
       source_sha256: typeof row.source_sha256 === "string" ? row.source_sha256
         : createHash("sha256").update(JSON.stringify(row)).digest("hex")});
   }
diff --git a/tests/control_plane/test_effective_turn_replan.py b/tests/control_plane/test_effective_turn_replan.py
index 57cd117791..e64a28ca7a 100644
--- a/tests/control_plane/test_effective_turn_replan.py
+++ b/tests/control_plane/test_effective_turn_replan.py
@@ -559,6 +559,31 @@ def call(*args: str, expected_code: int = 0) -> dict:
     call("quota", "spend-slot", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
          "--replan-obligation-id", next_obligation["obligation_id"], "--turn-instance-id", "stride-next-review",
          "--slots", "1", "--source", "heartbeat", "--execute")
+    # An unbound observation may name the declared work without claiming a Turn
+    # settlement. The public CLI rejects surface-only Turn settlement, but this
+    # ordinary status route must not consume the cited negative result either.
+    status_update = call("refresh-state", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+        "--progress-work-item-id", next_successor["todo_id"], "--progress-result-class", "unchanged",
+        "--classification", "status_readback",
+        "--delivery-batch-scale", "single_surface", "--delivery-outcome", "surface_only",
+        "--recommended-action", "The next probe remains planned; no new experiment has run.",
+        "--vision-summary", "The accepted source comparison is still pending; this is a status clarification only.",
+        "--no-global-sync", "--suppress-external-sinks")
+    status_run = json.loads(Path(status_update["json_path"]).read_text())
+    assert status_run["delivery_outcome"] == "surface_only"
+    assert status_run["agent_vision"], "the ordinary source must enter the causal history replay"
+    from loopx.control_plane.work_items.replan_context_codec import project_replan_context
+    from loopx.history import load_index_snapshot
+
+    context = project_replan_context(goal_id=GOAL_ID, agent_id=AGENT_ID,
+        runs=load_index_snapshot(runtime / "goals" / GOAL_ID / "runs/index.jsonl",
+                                 include_artifact_status=False).records,
+        history_newest_first=False, stride_strategy="evidence_conditioned_v1")
+    assert any(row["coverage_ref"] == status_run["progress_observation"]["fingerprint"]
+               for row in context["decision_followthrough"]["observations"])
+    pending = context["decision_followthrough"]["work_result_use"]
+    assert any(row["evidence_ref"] == ordinary_ref for row in pending["unresolved"])
+    assert not any(row["result_evidence_ref"] == ordinary_ref for row in pending["recorded_followthrough"])
     for index in range(2):
         turn = f"adopted-probe-{index}"
         guard = call("quota", "should-run", "--codex-app", "--goal-id", GOAL_ID,
diff --git a/tests/control_plane/test_research_execution_loop.py b/tests/control_plane/test_research_execution_loop.py
new file mode 100644
index 0000000000..c46a97c444
--- /dev/null
+++ b/tests/control_plane/test_research_execution_loop.py
@@ -0,0 +1,259 @@
+"""Synthetic real-process/CLI evidence for stage, candidate and result continuity."""
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+from types import SimpleNamespace
+
+import pytest
+
+from benchmark.runtime.research_loop import prepare_research_continuation
+from benchmark.runtime.research_probe import run_probe
+from canonical_authority_fixture import isolate_sqlite_runtime
+import test_quota_settlement_cli as fixture
+
+
+def probe(tmp_path, monkeypatch, *, source=None, **options):
+    monkeypatch.chdir(tmp_path)
+    candidate = tmp_path / "candidate.py"
+    candidate.write_text(source or "import json,os; from pathlib import Path; "
+        "Path(os.environ['LOOPX_RESEARCH_DIAGNOSTICS']).write_text(json.dumps("
+        "{'search_iterations': 2, 'stop_reason': 'search_cap'})); print(sum(x*x for x in (1,2)))")
+    return run_probe(command=[sys.executable, str(candidate)], candidate_files=[candidate],
+        output=tmp_path / "observation.json", input_revision="synthetic-input-v1",
+        planned_method_id="scheduler", executed_method_id="mask-selection",
+        change_reason="The measured control motivated a different first intervention.",
+        evidence_ref="probe:local-1", limit_seconds=options.pop("limit_seconds", 30),
+        diagnostics=tmp_path / "diagnostics.json", **options)
+
+
+def test_real_probe_reports_early_exit_not_budget_exhaustion(tmp_path, monkeypatch):
+    observation = probe(tmp_path, monkeypatch)
+    assert observation["search_iterations"] == 2
+    assert 0 < observation["elapsed_seconds"] < observation["limit_seconds"]
+    assert observation["stop_reason"] == "process_exit:0; solver_report:search_cap"
+    assert len(observation["candidate_sha256"]) == 64
+    assert (tmp_path / "observation.json.stdout").read_text().strip() == "5"
+
+
+def test_missing_solver_telemetry_stays_unknown_and_timeout_is_host_observed(tmp_path, monkeypatch):
+    observation = probe(tmp_path, monkeypatch, source="import time; time.sleep(10)", limit_seconds=0.05)
+    assert observation["search_iterations"] is None
+    assert observation["stop_reason"] == "host_timeout"
+    assert observation["elapsed_seconds"] >= 0.05
+
+
+def test_probe_refuses_changed_candidate_and_never_reuses_existing_evidence(tmp_path, monkeypatch):
+    feedback_root = tmp_path / "feedback"
+    monkeypatch.setenv("LOOPX_RESEARCH_FEEDBACK_ROOT", str(feedback_root))
+    with pytest.raises(ValueError, match="candidate changed"):
+        probe(tmp_path, monkeypatch, source="from pathlib import Path; Path(__file__).write_text('print(0)')")
+    before = (tmp_path / "observation.json").read_bytes()
+    with pytest.raises(FileExistsError):
+        probe(tmp_path, monkeypatch)
+    assert (tmp_path / "observation.json").read_bytes() == before
+    assert not feedback_root.exists(), "a changed/unexecuted candidate cannot register as a completed probe"
+
+
+@pytest.mark.parametrize("reported", [
+    "{",
+    '{"search_iterations":"unknown","stop_reason":"search_cap"}',
+    '{"search_iterations":2,"stop_reason":{"unexpected":"value"}}',
+    '{"unexpected":"do-not-retain-raw-diagnostic-body"}',
+    "x" * 4097,
+])
+def test_invalid_optional_diagnostics_preserve_host_measurements_and_bounded_error_evidence(tmp_path, monkeypatch, reported):
+    observed = probe(tmp_path, monkeypatch, source=(
+        "import os; from pathlib import Path; "
+        f"Path(os.environ['LOOPX_RESEARCH_DIAGNOSTICS']).write_text({reported!r}); "
+        "print('work completed'); raise SystemExit(3)"))
+    assert observed["elapsed_seconds"] > 0
+    assert observed["exit_code"] == 3 and observed["stop_reason"] == "process_exit:3"
+    assert observed["search_iterations"] is None
+    assert (tmp_path / "observation.json.stdout").read_text().strip() == "work completed"
+    assert json.loads((tmp_path / "observation.json").read_text()) == observed
+    raw_error = (tmp_path / "observation.json.diagnostics-error.json").read_bytes()
+    assert len(raw_error) < 1024 and b"do-not-retain-raw-diagnostic-body" not in raw_error
+    detail = json.loads(raw_error)
+    assert detail["source"] == "solver_diagnostics"
+    assert detail["candidate_sha256"] == observed["candidate_sha256"]
+    assert "diagnostics-error:" + hashlib.sha256(raw_error).hexdigest() in observed["evidence_refs"]
+
+
+def test_invalid_solver_report_cannot_hide_a_host_timeout(tmp_path, monkeypatch):
+    observed = probe(tmp_path, monkeypatch, source=(
+        "import os,time; from pathlib import Path; "
+        "Path(os.environ['LOOPX_RESEARCH_DIAGNOSTICS']).write_text('{'); time.sleep(10)"), limit_seconds=1)
+    assert observed["stop_reason"] == "host_timeout" and observed["exit_code"] < 0
+    assert observed["search_iterations"] is None and observed["elapsed_seconds"] >= 1
+    assert (tmp_path / "observation.json.diagnostics-error.json").is_file()
+
+
+def test_probe_cli_preserves_a_failed_candidate_exit_and_its_observation(tmp_path):
+    candidate = tmp_path / "candidate.py"
+    candidate.write_text("raise SystemExit(3)")
+    output = tmp_path / "observed.json"
+    result = subprocess.run([sys.executable, "-m", "benchmark.runtime.research_probe",
+        "--candidate-file", str(candidate), "--output", str(output), "--input-revision", "fixture-v1",
+        "--planned-method-id", "baseline", "--executed-method-id", "baseline",
+        "--evidence-ref", "probe:failed", "--limit-seconds", "5", "--", sys.executable, str(candidate)],
+        cwd=tmp_path, env={**os.environ, "PYTHONPATH": str(fixture.REPO_ROOT)}, capture_output=True, text=True)
+    assert result.returncode == 3, result.stderr
+    observed = json.loads(output.read_text())
+    assert observed["exit_code"] == 3 and observed["search_iterations"] is None
+    assert json.loads(output.with_suffix(".json.candidate.json").read_text())["candidate_sha256"] == observed["candidate_sha256"]
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_candidate_stage_leaves_research_work_and_execution_reaches_next_context(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = fixture._write_fixture(tmp_path)
+    configuration = json.loads(registry.read_text())
+    configuration["goals"][0]["quota"]["allowed_slots"] = 12
+    registry.write_text(json.dumps(configuration))
+    fixture._initialize_fixture_authority(project, runtime, provider=provider)
+    monkeypatch.chdir(project)
+
+    def call(*args, expected=0):
+        code, result = fixture._run_cli(registry, runtime, *args, cwd=project)
+        assert code == expected, json.dumps({key: result.get(key) for key in (
+            "ok", "error", "error_code", "effective_action", "reason", "blocker", "autonomous_replan_obligation")})
+        return result
+
+    async def cli(_environment, args, **_kwargs):
+        return call(*args)
+
+    call("configure-goal", "--goal-id", fixture.GOAL_ID, "--execution-replan-after-turns", "2",
+        "--execution-stride-strategy", "evidence_conditioned_v1", "--execute")
+    owner = SimpleNamespace(_seeded_todo_id=fixture.TODO_ID, _loopx=cli,
+        goal_id=fixture.GOAL_ID, agent_id=fixture.AGENT_ID, _task_document="TASK.md")
+    root = asyncio.run(prepare_research_continuation(owner, None, cwd=str(project)))
+    original = registry.read_bytes()
+    again = asyncio.run(prepare_research_continuation(owner, None, cwd=str(project)))
+    assert again["todo_id"] == root["todo_id"]
+    assert registry.read_bytes() == original
+    call("todo", "complete", "--goal-id", fixture.GOAL_ID, "--agent-id", fixture.AGENT_ID,
+         "--todo-id", fixture.TODO_ID)
+    binding = ("--goal-id", fixture.GOAL_ID, "--agent-id", fixture.AGENT_ID)
+    current = call("quota", "should-run", *binding, "--runtime-profile", "generic_cli",
+        "--turn-instance-id", "research-after-candidate", "--scan-path", str(project))
+    assert current["should_run"]
+    assert current["selected_todo"]["todo_id"] == root["todo_id"]
+    assert current.get("goal_terminal_state") is None
+    assert current["effective_action"] == "successor_replan_required"
+    # Follow the existing resume action. Readiness is not itself a durable
+    # open-state transition and must not be treated as completed research.
+    call("todo", "update", *binding, "--todo-id", root["todo_id"], "--status", "open",
+        "--clear-resume-when", "--reason", "The initial candidate stage completed; perform the retained research work.")
+    scope = project / "research-scope.json"
+    scope.write_text(json.dumps({"schema_version": "goal_vision_replan_contract_v0", "state": "active",
+        "vision_patch": {"advancement_policy": "as_needed", "vision_summary": "The retained research work is now runnable.",
+            "acceptance_summary": "Compare actual mechanisms and resources against the original objective."},
+        "path_delta": {"schema_version": "goal_path_delta_v0", "outcome": "continue",
+            "prior_assumption": "Research work awaits the initial candidate stage.",
+            "observed_reality": "The canonical candidate stage is done and its retained research successor is open.",
+            "retained": ["Original objective, resource limits and best candidate."],
+            "changed": ["Run the retained research work."], "evidence_refs": ["todo:" + fixture.TODO_ID]}}))
+    preparation = call("refresh-state", *binding, "--todo-id", root["todo_id"], "--turn-instance-id", "research-after-candidate",
+        "--classification", "successor_replan_recorded", "--delivery-batch-scale", "implementation",
+        "--delivery-outcome", "outcome_progress", "--agent-vision-json", str(scope),
+        "--no-global-sync", "--suppress-external-sinks")
+    assert json.loads(Path(preparation["json_path"]).read_text())["vision_checkpoint"]["satisfied"]
+    call("quota", "spend-slot", *binding, "--todo-id", root["todo_id"], "--turn-instance-id", "research-after-candidate",
+        "--slots", "1", "--source", "heartbeat", "--execute")
+    work_turn = "research-local-probe"
+    work_guard = call("quota", "should-run", *binding, "--runtime-profile", "generic_cli",
+        "--todo-id", root["todo_id"], "--turn-instance-id", work_turn, "--scan-path", str(project))
+    assert work_guard["should_run"] and work_guard["effective_action"] == "normal_run"
+    call("configure-goal", "--goal-id", fixture.GOAL_ID, "--explore-graph-enabled", "--execute")
+    observed = probe(project, monkeypatch)
+    observed.update(result_ref="result:released-1", result_candidate_sha256=observed["candidate_sha256"])
+    observed["evidence_refs"].append("result:released-1")
+    observed["evidence_refs"].sort()
+    attachment = {"schema_version": "explore_result_attachment_v0", "question": "Which mechanism improves the bounded case?",
+        "applicability": "Synthetic public local inputs only", "input_revision": observed["input_revision"],
+        "observation": "The process stopped after two search steps with allowance remaining.",
+        "interpretation": "Retain the incumbent; change the search cap before claiming scheduler benefit.",
+        "status": "tentative", "evidence_refs": observed["evidence_refs"], "execution_observation": observed}
+    path = project / "result.json"
+    path.write_text(json.dumps(attachment))
+    result = call("refresh-state", *binding, "--todo-id", root["todo_id"],
+        "--turn-instance-id", work_turn, "--classification", "validated_probe",
+        "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
+        "--vision-unchanged-reason", "Original research remains open; the changed intervention needs validation.",
+        "--explore-result-json", str(path), "--no-global-sync", "--suppress-external-sinks")
+    assert result["explore_result_delivery"]["ok"]
+    context = call("explore", "turn-context", *binding)
+    [finding] = context["graph"]["writeback_results"]
+    assert finding["execution_observation"] == observed
+    from loopx.capabilities.explore.result_log import explore_result_log_path, load_explore_result_events_strict, explore_history_rows
+    rows = explore_history_rows(load_explore_result_events_strict(explore_result_log_path(runtime, fixture.GOAL_ID),
+                                goal_id=fixture.GOAL_ID), goal_id=fixture.GOAL_ID)
+    assert next(row for row in rows if row["event_kind"] == "finding")["execution_observation"] == observed
+    # Late feedback for another snapshot must not be attached to this execution.
+    attachment["execution_observation"]["result_candidate_sha256"] = "b" * 64
+    path.write_text(json.dumps(attachment))
+    bad = call("refresh-state", *binding, "--todo-id", root["todo_id"],
+        "--turn-instance-id", work_turn, "--classification", "validated_probe",
+        "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
+        "--explore-result-json", str(path), "--no-global-sync", "--suppress-external-sinks", expected=1)
+    assert not bad.get("appended")
+    call("quota", "spend-slot", *binding, "--todo-id", root["todo_id"],
+        "--turn-instance-id", work_turn, "--slots", "1", "--source", "heartbeat", "--execute")
+    review = call("quota", "should-run", *binding, "--runtime-profile", "generic_cli",
+        "--turn-instance-id", "review-resource-evidence", "--scan-path", str(project))
+    assert "replan_action_packet" in review, {key: review.get(key) for key in (
+        "should_run", "effective_action", "reason", "autonomous_replan_obligation", "effective_turn_replan")}
+    control = review["replan_action_packet"]["stride_control"]
+    delivered = next(item for item in control["explore_basis"]["finding_refs"] if item["finding_id"] == finding["finding_id"])
+    assert delivered["execution_observation"]["candidate_sha256"] == observed["candidate_sha256"]
+    assert delivered["execution_observation"]["search_iterations"] == 2
+    # The actor uses delivered measurements, then runs a changed implementation;
+    # a new finding/identifier alone is not the observable outcome.
+    assert delivered["execution_observation"]["elapsed_seconds"] < delivered["execution_observation"]["limit_seconds"]
+    next_work = call("todo", "add", "--goal-id", fixture.GOAL_ID, "--role", "agent",
+        "--claimed-by", fixture.AGENT_ID, "--task-class", "advancement_task", "--action-kind", "validate",
+        "--target-key", "search-cap-control", "--replan-obligation-id", review["autonomous_replan_obligation"]["obligation_id"],
+        "--text", "Vary the measured search cap while retaining the incumbent and the original method claim.")
+    assert next_work["todo_id"] != root["todo_id"]
+    review = call("quota", "should-run", *binding, "--runtime-profile", "generic_cli",
+        "--todo-id", next_work["todo_id"], "--turn-instance-id", "review-resource-evidence", "--scan-path", str(project))
+    control = review["replan_action_packet"]["stride_control"]
+    decision = control["minimal_example"]
+    decision["explore_review"] = {"revision": control["explore_basis"]["revision"],
+        "disposition": "used", "considered_refs": [finding["finding_id"]],
+        "reason": "The executed mask candidate stopped at its search cap, not at the resource limit."}
+    path.write_text(json.dumps({"schema_version": "goal_vision_replan_contract_v0", "state": "vision_patch_proposed",
+        "vision_patch": {"acceptance_summary": "Validate a changed search cap against the retained reference."},
+        "path_delta": {"schema_version": "goal_path_delta_v0", "outcome": "replan",
+            "prior_assumption": "A new scheduler was expected to explain the gain.",
+            "observed_reality": "The executed artifact uses mask selection and stops after two search steps.",
+            "changed": ["Vary the search cap before attributing gain to a scheduler."],
+            "evidence_refs": [finding["finding_id"]]},
+        "stride_decision": decision, "todo_delta": ["create:" + next_work["todo_id"]]}))
+    replan_binding = (*binding, "--replan-obligation-id", review["autonomous_replan_obligation"]["obligation_id"],
+                      "--turn-instance-id", "review-resource-evidence")
+    accepted = call("refresh-state", *replan_binding, "--classification", "resource_probe_replan",
+        "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
+        "--agent-vision-json", str(path), "--no-global-sync", "--suppress-external-sinks")
+    assert accepted["autonomous_replan_ack"]["semantic_delta"]["accepted"]
+    call("quota", "spend-slot", *replan_binding, "--slots", "1", "--source", "heartbeat", "--execute")
+    admitted = call("quota", "should-run", *binding, "--runtime-profile", "generic_cli",
+        "--todo-id", next_work["todo_id"], "--turn-instance-id", "execute-search-control", "--scan-path", str(project))
+    assert admitted["should_run"] and admitted["selected_todo"]["todo_id"] == next_work["todo_id"]
+    candidate = project / "candidate.py"
+    candidate.write_text("import json,os; from pathlib import Path; "
+        "Path(os.environ['LOOPX_RESEARCH_DIAGNOSTICS']).write_text(json.dumps("
+        "{'search_iterations': 5, 'stop_reason': 'local_cases_complete'})); print(1)")
+    revised = run_probe(command=[sys.executable, str(candidate)], candidate_files=[candidate],
+        output=project / "next-observation.json", input_revision="synthetic-input-v1",
+        planned_method_id="mask-selection", executed_method_id="mask-selection", evidence_ref="probe:local-2",
+        limit_seconds=30, diagnostics=project / "next-diagnostics.json")
+    assert revised["candidate_sha256"] != observed["candidate_sha256"]
+    assert revised["search_iterations"] == 5
+    assert (project / "next-observation.json.stdout").read_text().strip() == "1"
diff --git a/tests/control_plane_ts/explore_result_writeback.test.ts b/tests/control_plane_ts/explore_result_writeback.test.ts
index 3a53311e55..031d80d167 100644
--- a/tests/control_plane_ts/explore_result_writeback.test.ts
+++ b/tests/control_plane_ts/explore_result_writeback.test.ts
@@ -11,6 +11,31 @@ const attachment = {
   evidence_refs: ["validation:prefix", "validation:tail"],
 };
 
+const execution = {schema_version: "explore_execution_observation_v0",
+  candidate_sha256: "a".repeat(64), input_revision: attachment.input_revision,
+  planned_method_id: "scheduler", executed_method_id: "mask-selection",
+  change_reason: "The control identified mask selection as the first discriminating change.",
+  elapsed_seconds: 0.2, limit_seconds: 30, search_iterations: 2,
+  stop_reason: "process_exit:0; solver_report:search_cap", evidence_refs: attachment.evidence_refs};
+
+test("execution observations bind resources and changed methods without implying adoption", () => {
+  const normalized = normalizeExploreResultAttachment({attachment: {...attachment, execution_observation: execution}});
+  assert.deepEqual(normalized.execution_observation, execution);
+  assert.deepEqual(normalizeExploreResultAttachment({attachment}), attachment);
+  const unknown = {...execution, elapsed_seconds: null, search_iterations: null, limit_seconds: null, stop_reason: null};
+  assert.deepEqual(normalizeExploreResultAttachment({attachment: {...attachment, execution_observation: unknown}}).execution_observation, unknown);
+  for (const mutation of [{change_reason: null}, {input_revision: "foreign-input"},
+    {search_iterations: -1}, {search_iterations: 1.1}, {elapsed_seconds: NaN}, {limit_seconds: 0},
+    {evidence_refs: ["foreign:probe"]}, {candidate_sha256: "finding-digest-is-not-code"},
+    {result_ref: attachment.evidence_refs[0]},
+    {result_ref: attachment.evidence_refs[0], result_candidate_sha256: "b".repeat(64)}]) {
+    assert.throws(() => normalizeExploreResultAttachment({attachment: {...attachment,
+      execution_observation: {...execution, ...mutation}}}));
+  }
+  const result = {...execution, result_ref: attachment.evidence_refs[0], result_candidate_sha256: execution.candidate_sha256};
+  assert.deepEqual(normalizeExploreResultAttachment({attachment: {...attachment, execution_observation: result}}).execution_observation, result);
+});
+
 test("question lifecycle is explicit, independent of finding status, and available on path capture", () => {
   const update = {status: "resolved"};
   const relations = [{to_node: "uniform-estimate", edge_type: "depends_on", summary: "Required before transfer."}];
diff --git a/tests/control_plane_ts/replan_context.test.ts b/tests/control_plane_ts/replan_context.test.ts
index 2dd917d11a..e7413a2259 100644
--- a/tests/control_plane_ts/replan_context.test.ts
+++ b/tests/control_plane_ts/replan_context.test.ts
@@ -134,7 +134,7 @@ const strideRequest = {...request, stride_strategy: "evidence_conditioned_v1"};
 test("unconsumed result survives later reviews and citations alone; recorded work closes only its source relation", () => {
   const first = reviewed(1);
   (first._source_facts as JsonObject).planned_todo_ids = ["todo-a"];
-  const result = {...subsequent(2), progress_observation: null};
+  const result = {...subsequent(2), progress_observation: null, delivery_outcome: "outcome_progress"};
   const ref = (projectReplanContext({...request, rows: [result]}).evidence as JsonObject[])[0].evidence_ref;
   const next = reviewed(3);
   (next._source_facts as JsonObject).planned_todo_ids = ["todo-b"];
@@ -146,7 +146,18 @@ test("unconsumed result survives later reviews and citations alone; recorded wor
   assert.equal(proposed.unresolved_count, 1);
   assert.equal((proposed.unresolved as JsonObject[])[0].evidence_ref, ref);
   assert.deepEqual(proposed.recorded_followthrough, []);
-  const actual = {...subsequent(4), todo_id: "todo-b", progress_observation: null};
+  // A later record is not necessarily work. Keep preparation and unknown
+  // legacy outcomes readable without consuming the result or creating a new
+  // pending research result from each status update.
+  for (const delivery_outcome of ["surface_only", null, "future_outcome"]) {
+    for (const progress_observation of [null, {...progress, work_item_id: "todo-b"}]) {
+      const status = {...subsequent(4), todo_id: "todo-b", delivery_outcome, progress_observation};
+      assert.deepEqual(readUse([first, result, next, status]), proposed);
+      assert.equal((projectReplanContext({...request, rows: [status]}).evidence as JsonObject[]).length, 1);
+    }
+  }
+  const actual = {...subsequent(4), todo_id: "todo-b", progress_observation: null,
+    delivery_outcome: "outcome_progress"};
   const adopted = readUse([first, result, next, actual]);
   assert.equal(adopted.unresolved_count, 1, "the next result remains available for the next decision");
   assert.notEqual((adopted.unresolved as JsonObject[])[0].evidence_ref, ref);
@@ -156,6 +167,15 @@ test("unconsumed result survives later reviews and citations alone; recorded wor
   assert.deepEqual(readUse([first, result, next, actual, actual]), adopted, "replay cannot consume or duplicate twice");
   const replayedResult = {...result, _source_facts: {...result._source_facts as JsonObject, history_position: 5}};
   assert.deepEqual(readUse([first, result, next, actual, replayedResult]), adopted, "a lost-response replay cannot resurrect consumed evidence");
+  // A negative experiment or a new concrete blocker can still inform a later
+  // decision. This relation must not require a positive scientific outcome.
+  for (const delivery_outcome of ["outcome_gap", "primary_goal_outcome"]) {
+    const use = readUse([first, result, next, {...actual, delivery_outcome}]);
+    assert.equal(use.recorded_followthrough_count, 1);
+    assert.equal((use.recorded_followthrough as JsonObject[])[0].result_evidence_ref, ref);
+  }
+  const metadata = {...result, delivery_outcome: "surface_only"};
+  assert.equal(readUse([first, metadata]), undefined, "metadata is not a new unresolved work result");
   const omitted = reviewed(3);
   (omitted._source_facts as JsonObject).planned_todo_ids = ["todo-b"];
   assert.equal(readUse([first, result, omitted]).unresolved_count, 1, "another review cannot silently drop an uncited result");