From c66ece1790a1da680cc9a673122dc16221d4bf90 Mon Sep 17 00:00:00 2001
From: shangzh0 <97216392+shangzh0@users.noreply.github.com>
Date: Sat, 31 Oct 2026 19:36:19 +0000
Subject: [PATCH 1/3] fix(todo): retire old next steps after explicit guidance

Signed-off-by: shangzh0 <97216392+shangzh0@users.noreply.github.com>
---
 docs/project-agent-todo-contract.md           |  5 ++++
 .../work_items/recommendation_source_io.py    | 10 +++++++-
 .../test_next_action_writeback.py             | 24 +++++++++++++++++++
 3 files changed, 38 insertions(+), 1 deletion(-)

diff --git a/docs/project-agent-todo-contract.md b/docs/project-agent-todo-contract.md
index 1c06f8d9ae..d3007311d4 100644
--- a/docs/project-agent-todo-contract.md
+++ b/docs/project-agent-todo-contract.md
@@ -591,6 +591,11 @@ Todo, in either personal or Goal report scope; it does not replace shared prose.
 `--next-action-basis` optionally rejects an edit planned from an older source
 read. If the basis is stale, read the current task/step and retry; if no eligible
 task is available, resolve its existing lifecycle or routing boundary first.
+A later explicit `--recommended-action` from that same Agent retires reuse of
+its older bound step. The selected Todo and its authority remain unchanged;
+another Agent's guidance does not retire the step, and ordinary derived progress
+observations preserve it. Write a fresh `--next-action` if a new task-bound step
+is needed after the explicit guidance.
 Native Turn host `next_action` remains follow-up guidance in its durable result,
 not this explicit task-step edit: completion or repair may leave no runnable
 current task. See [Next Action writeback](quota-allocation.md) for attribution,
diff --git a/loopx/control_plane/work_items/recommendation_source_io.py b/loopx/control_plane/work_items/recommendation_source_io.py
index 190193b8c8..96d0e4fe29 100644
--- a/loopx/control_plane/work_items/recommendation_source_io.py
+++ b/loopx/control_plane/work_items/recommendation_source_io.py
@@ -66,9 +66,17 @@ def recommendation_runs(runtime_root: Path, goal_id: str) -> list[dict[str, Any]
 
 def latest_bound_recommendation(runs: list[dict[str, Any]], agent_id: str) -> dict[str, Any] | None:
     # Journal decoding only. The TS owner validates source, actor and task binding.
+    from .refresh_recommendation import RECOMMENDED_ACTION_SOURCE_EXPLICIT
+
     for row in runs:
         resolution = row.get("recommended_action_resolution")
-        if row.get("agent_id") == agent_id and isinstance(resolution, dict) and resolution.get("step_revision"):
+        if row.get("agent_id") == agent_id and isinstance(resolution, dict) and (
+            resolution.get("step_revision")
+            or resolution.get("recommended_action_source") == RECOMMENDED_ACTION_SOURCE_EXPLICIT
+        ):
+            # Preserve an explicit newer decision as the read barrier. Ordinary
+            # derived observations still reuse the original step receipt; they
+            # do not mint a revision or revive a step before this decision.
             return resolution
     return None
 
diff --git a/tests/control_plane/test_next_action_writeback.py b/tests/control_plane/test_next_action_writeback.py
index 7295acb5ad..2c9076e6ce 100644
--- a/tests/control_plane/test_next_action_writeback.py
+++ b/tests/control_plane/test_next_action_writeback.py
@@ -109,6 +109,30 @@ def test_multi_agent_steps_are_independent_and_report_scope_does_not_grant_autho
     assert state.read_text() == original_state
 
 
+def test_explicit_guidance_retires_the_old_step_without_changing_the_selected_todo(tmp_path):
+    registry, state, runtime, _ = fixture(tmp_path, ("agent-a", "agent-b"))
+    original = state.read_text()
+    write(registry, runtime, next_action="Continue the old experiment.")
+    # A derived progress observation is not a new decision and keeps that step.
+    write(registry, runtime, next_action=None)
+    _, retained = routes(registry, runtime, tmp_path)
+    assert retained["agent-a"]["next_step"] == "Continue the old experiment."
+    # Another actor's explicit instruction cannot retire this actor's step.
+    write(registry, runtime, agent_id="agent-b", next_action=None,
+          recommended_action="Review the independent result.")
+    _, peer = routes(registry, runtime, tmp_path)
+    assert peer["agent-a"]["next_step"] == "Continue the old experiment."
+    decision = write(registry, runtime, next_action=None,
+                     recommended_action="The result invalidated the old experiment; reconsider the method.")
+    assert decision["recommended_action_resolution"]["recommended_action_source"] == "explicit_arg"
+    _, current = routes(registry, runtime, tmp_path)
+    assert current["agent-a"]["todo_id"] == "todo_parser"
+    assert "next_step" not in current["agent-a"], "a prior step cannot outrank later explicit guidance"
+    derived = write(registry, runtime, next_action=None)
+    assert derived["recommended_action"] != "Continue the old experiment."
+    assert state.read_text() == original
+
+
 @pytest.mark.parametrize("agents,actor", [([], "agent-a"), (["agent-b"], "agent-a"), (["agent-a"], None)])
 def test_unknown_or_unattributed_steps_do_not_append(tmp_path, agents, actor):
     registry, state, runtime, _ = fixture(tmp_path, agents)

From 2baba90005466298b89735cee30791768ff971c6 Mon Sep 17 00:00:00 2001
From: shangzh0 <97216392+shangzh0@users.noreply.github.com>
Date: Sat, 31 Oct 2026 20:26:39 +0000
Subject: [PATCH 2/3] fix(todo): retain explicit decision freshness in the read
 basis

Signed-off-by: shangzh0 <97216392+shangzh0@users.noreply.github.com>
---
 docs/project-agent-todo-contract.md           | 11 ++++
 .../work_items/recommendation_source_io.py    |  5 ++
 .../work_items/refresh_recommendation.ts      |  8 ++-
 .../test_next_action_writeback.py             | 52 ++++++++++++++++++-
 .../refresh_recommendation.test.ts            | 23 ++++++++
 5 files changed, 96 insertions(+), 3 deletions(-)

diff --git a/docs/project-agent-todo-contract.md b/docs/project-agent-todo-contract.md
index d3007311d4..1bf2a37150 100644
--- a/docs/project-agent-todo-contract.md
+++ b/docs/project-agent-todo-contract.md
@@ -596,6 +596,17 @@ its older bound step. The selected Todo and its authority remain unchanged;
 another Agent's guidance does not retire the step, and ordinary derived progress
 observations preserve it. Write a fresh `--next-action` if a new task-bound step
 is needed after the explicit guidance.
+
+The same Agent's explicit retirement also remains part of the Next Action read
+fence. The typed owner hashes its existing journal source identity, so a delayed
+initial basis cannot become valid again, and repeated wording still names distinct
+committed decisions. This is an ephemeral source-facts fence, not a new persisted
+step or an execution grant; a fresh read can establish the next valid step.
+
+同一 Agent 的显式步骤退休也参与 Next Action 读取依据。typed owner 对已有 journal
+来源身份计算摘要，旧的初始依据不能再次有效；即使指导文字重复，独立提交的决策也不
+混为同一依据。这不是新增持久化步骤或执行权限，新读取仍可合法写入下一步。
+
 Native Turn host `next_action` remains follow-up guidance in its durable result,
 not this explicit task-step edit: completion or repair may leave no runnable
 current task. See [Next Action writeback](quota-allocation.md) for attribution,
diff --git a/loopx/control_plane/work_items/recommendation_source_io.py b/loopx/control_plane/work_items/recommendation_source_io.py
index 96d0e4fe29..1bf457cca9 100644
--- a/loopx/control_plane/work_items/recommendation_source_io.py
+++ b/loopx/control_plane/work_items/recommendation_source_io.py
@@ -77,6 +77,11 @@ def latest_bound_recommendation(runs: list[dict[str, Any]], agent_id: str) -> di
             # Preserve an explicit newer decision as the read barrier. Ordinary
             # derived observations still reuse the original step receipt; they
             # do not mint a revision or revive a step before this decision.
+            if resolution.get("recommended_action_source") == RECOMMENDED_ACTION_SOURCE_EXPLICIT:
+                # Existing journal identity facts, not a new persisted decision
+                # or step. The TS owner binds them into the read fence only.
+                return {**resolution, "agent_id": row["agent_id"], "goal_id": row.get("goal_id"),
+                    "journal_source": {key: row.get(key) for key in ("generated_at", "json_path")}}
             return resolution
     return None
 
diff --git a/loopx/control_plane/work_items/refresh_recommendation.ts b/loopx/control_plane/work_items/refresh_recommendation.ts
index b8deaa06af..d51fac282a 100644
--- a/loopx/control_plane/work_items/refresh_recommendation.ts
+++ b/loopx/control_plane/work_items/refresh_recommendation.ts
@@ -83,8 +83,12 @@ export function resolveLaneRecommendation(value: unknown): JsonObject {
   const prior = request.prior_resolution == null ? null : requireJsonObject(request.prior_resolution, "prior_resolution");
   // Only the actor's latest bound receipt participates. Never search older
   // matching tasks: doing so would revive an abandoned experiment.
-  const priorRevision = prior?.recommended_action_source === "agent_lane_step" && prior.agent_id === actor
-    ? prior.step_revision : null;
+  const priorRevision = prior?.agent_id !== actor ? null
+    : prior?.recommended_action_source === "agent_lane_step" ? prior.step_revision
+    // Retiring a step is a later decision, never a return to the initial basis.
+    // Hash the existing explicit journal receipt (including its source identity)
+    // without making it a reusable step or a new task/execution authority.
+    : prior?.recommended_action_source === "explicit_arg" && prior.goal_id === goalId ? digest(prior) : null;
   const basis = digest([goalId, sourceRevision, agents, actor, taskBasis, priorRevision]);
   const runnable = actor !== null && agents.includes(actor) && selected !== null &&
     selected.task_class === "advancement_task" && candidateIsRunnable(selected, actor);
diff --git a/tests/control_plane/test_next_action_writeback.py b/tests/control_plane/test_next_action_writeback.py
index 2c9076e6ce..198701f07e 100644
--- a/tests/control_plane/test_next_action_writeback.py
+++ b/tests/control_plane/test_next_action_writeback.py
@@ -8,7 +8,7 @@
 
 import loopx.state_refresh as refresh
 from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority
-from loopx.control_plane.testing.canary_harness import run_json_cli
+from loopx.control_plane.testing.canary_harness import run_json_cli, run_json_cli_result
 from loopx.control_plane.todos.provider_projection import project_current_canonical_todos
 from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
 from loopx.control_plane.work_items.recommendation_source_io import RecommendationWritebackRejected
@@ -353,3 +353,53 @@ def unavailable(**kwargs):
         write(registry, runtime)
     assert state.read_text() == original_state
     assert not (runtime / "goals/next-action-goal/runs/index.jsonl").exists()
+
+
+def test_explicit_retirement_cannot_revalidate_a_delayed_initial_basis(tmp_path):
+    registry, state, runtime, _ = fixture(tmp_path)
+    original_state = state.read_bytes()
+    authority = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="next-action-goal")
+    _, initial = routes(registry, runtime, tmp_path)
+    old_basis = initial["agent-a"]["next_action_basis"]
+    write(registry, runtime, next_action="Continue the old experiment.", next_action_basis=old_basis)
+    guidance = "The result invalidated the experiment; reconsider the method."
+    write(registry, runtime, next_action=None, recommended_action=guidance)
+    _, retired = routes(registry, runtime, tmp_path)
+    new_basis = retired["agent-a"]["next_action_basis"]
+    assert new_basis != old_basis, "an explicit decision must not recreate the initial read basis"
+    index = runtime / "goals/next-action-goal/runs/index.jsonl"
+    before = index.read_bytes()
+    with pytest.raises(RecommendationWritebackRejected) as conflict:
+        write(registry, runtime, next_action="Continue the old experiment.", next_action_basis=old_basis)
+    assert conflict.value.code == "next_action_basis_conflict"
+    assert index.read_bytes() == before
+    delayed = subprocess.run([sys.executable, "-m", "loopx.cli", "--format", "json",
+        "--registry", str(registry), "--runtime-root", str(runtime), "refresh-state",
+        "--goal-id", "next-action-goal", "--agent-id", "agent-a",
+        "--next-action", "Continue the old experiment.", "--next-action-basis", old_basis,
+        "--no-global-sync"], capture_output=True, text=True, timeout=30)
+    assert delayed.returncode == 1
+    assert json.loads(delayed.stdout)["error_code"] == "next_action_basis_conflict"
+    assert index.read_bytes() == before
+    status_code, status = run_json_cli_result("status", "--include-task-graph", registry_path=registry, runtime_root=runtime)
+    # This existing minimal fixture lacks a complete Goal health brief; status
+    # still returns a complete read projection, not an execution/health grant.
+    assert status_code == 1 and status["projection_envelope"]["complete"] is True
+    assert "error" not in status
+    item = next(item for item in status["attention_queue"]["items"] if item["goal_id"] == "next-action-goal")
+    selected = next(row for row in item["agent_next_actions"] if row["agent_id"] == "agent-a")
+    assert "next_step" not in selected
+    # Even a repeated explicit statement is a later decision, not the same read.
+    write(registry, runtime, next_action=None, recommended_action=guidance)
+    _, repeated = routes(registry, runtime, tmp_path)
+    assert repeated["agent-a"]["next_action_basis"] != new_basis
+    write(registry, runtime, next_action="Test the newly supported route.",
+        next_action_basis=repeated["agent-a"]["next_action_basis"])
+    for _ in range(2):
+        write(registry, runtime, next_action=None)
+    _, current = routes(registry, runtime, tmp_path)
+    assert current["agent-a"]["next_step"] == "Test the newly supported route."
+    assert state.read_bytes() == original_state
+    after = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="next-action-goal")
+    for key in ("provider_revision", "cursor", "todos"):
+        assert after[key] == authority[key]
diff --git a/tests/control_plane_ts/refresh_recommendation.test.ts b/tests/control_plane_ts/refresh_recommendation.test.ts
index 4315e4531c..8052965372 100644
--- a/tests/control_plane_ts/refresh_recommendation.test.ts
+++ b/tests/control_plane_ts/refresh_recommendation.test.ts
@@ -309,3 +309,26 @@ test("lane write revalidates the original Goal, actor and effect identity", () =
     ...baseRequest.settlement_identity, effect_id: "forged-effect",
   }}), /effect id does not match its identity/);
 });
+
+test("explicit retirement keeps a distinct read fence instead of recreating the initial basis", () => {
+  const initial = resolveLaneRecommendation(laneContext);
+  const first = resolveLaneRecommendation({...laneContext, write: {text: "Old experiment.", expected_basis: initial.basis}});
+  const explicit = {...resolveRefreshRecommendation({...baseRequest, explicit_action: "Reconsider the method."}),
+    goal_id: laneContext.goal_id, agent_id: laneContext.agent_id,
+    journal_source: {generated_at: "2026-01-01T00:00:00Z", json_path: "run:explicit-1"}};
+  const retired = resolveLaneRecommendation({...laneContext, prior_resolution: explicit});
+  assert.notEqual(retired.basis, initial.basis);
+  assert.notEqual(retired.basis, resolveLaneRecommendation({...laneContext, prior_resolution: first.resolution}).basis);
+  assert.equal((retired.selected_todo as Record<string, unknown>).next_step, undefined);
+  const stale = resolveLaneRecommendation({...laneContext, prior_resolution: explicit,
+    write: {text: "Old experiment.", expected_basis: initial.basis}});
+  assert.equal(stale.admitted, false);
+  assert.equal(stale.error_code, "next_action_basis_conflict");
+  assert.equal(resolveLaneRecommendation({...laneContext, prior_resolution: explicit,
+    write: {text: "New supported route.", expected_basis: retired.basis}}).admitted, true);
+  const repeated = resolveLaneRecommendation({...laneContext, prior_resolution: {...explicit,
+    journal_source: {...explicit.journal_source, json_path: "run:explicit-2"}}});
+  assert.notEqual(repeated.basis, retired.basis, "same wording and clock do not collapse distinct committed receipts");
+  assert.equal(resolveLaneRecommendation({...laneContext, prior_resolution: {...explicit, agent_id: "agent-b"}}).basis, initial.basis);
+  assert.equal(JSON.stringify(retired).includes("run:explicit"), false, "source identity stays behind the digest");
+});

From 2e3e102475726555f4e55d9174806e769bc69981 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 1 Nov 2026 10:24:39 +0800
Subject: [PATCH 3/3] fix(recommendation): preserve supersession in retained
 quota history

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/project-agent-todo-contract.md           |  5 ++
 .../goals/goal_frontier/semantic_history.py   |  6 +-
 loopx/control_plane/runtime/run_compaction.py |  9 ++-
 .../runtime/run_context_retention.py          |  8 ++-
 .../work_items/recommendation_source_io.py    | 32 +++++-----
 .../test_receipt_next_action_cli.py           | 63 +++++++++++++++++++
 .../test_run_context_retention.py             | 28 +++++++++
 7 files changed, 127 insertions(+), 24 deletions(-)

diff --git a/docs/project-agent-todo-contract.md b/docs/project-agent-todo-contract.md
index 1bf2a37150..4fc3d6b8eb 100644
--- a/docs/project-agent-todo-contract.md
+++ b/docs/project-agent-todo-contract.md
@@ -602,10 +602,15 @@ fence. The typed owner hashes its existing journal source identity, so a delayed
 initial basis cannot become valid again, and repeated wording still names distinct
 committed decisions. This is an ephemeral source-facts fence, not a new persisted
 step or an execution grant; a fresh read can establish the next valid step.
+Full and retained history preserve that same decision and source identity beyond
+the recent display window. Status and quota reads therefore provide the same
+current basis for the next step write.
 
 同一 Agent 的显式步骤退休也参与 Next Action 读取依据。typed owner 对已有 journal
 来源身份计算摘要，旧的初始依据不能再次有效；即使指导文字重复，独立提交的决策也不
 混为同一依据。这不是新增持久化步骤或执行权限，新读取仍可合法写入下一步。
+完整历史与保留历史在最近展示窗口之外也保留同一决策及其来源身份，status 与 quota
+读取因而为下一步写入提供一致的当前依据。
 
 Native Turn host `next_action` remains follow-up guidance in its durable result,
 not this explicit task-step edit: completion or repair may leave no runnable
diff --git a/loopx/control_plane/goals/goal_frontier/semantic_history.py b/loopx/control_plane/goals/goal_frontier/semantic_history.py
index 4e633fb4d0..3fccb64242 100644
--- a/loopx/control_plane/goals/goal_frontier/semantic_history.py
+++ b/loopx/control_plane/goals/goal_frontier/semantic_history.py
@@ -266,11 +266,11 @@ def latest_lane_recommendation_from_status(
     status_payload: dict[str, Any], *, goal_id: str, agent_id: str,
 ) -> dict[str, Any] | None:
     """Reuse the retained recommendation receipt beyond the recent-run window."""
+    from ...work_items.recommendation_source_io import latest_bound_recommendation
+
     present, context = _semantic_agent_context_for_goal(
         status_payload, goal_id=goal_id, agent_id=agent_id)
     if present:
         row = context.get("latest_lane_step_run") if isinstance(context, dict) else None
-        resolution = row.get("recommended_action_resolution") if isinstance(row, dict) else None
-        return resolution if isinstance(resolution, dict) else None
-    from ...work_items.recommendation_source_io import latest_bound_recommendation
+        return latest_bound_recommendation([row], agent_id) if isinstance(row, dict) else None
     return latest_bound_recommendation(_latest_runs_for_goal(status_payload, goal_id=goal_id), agent_id)
diff --git a/loopx/control_plane/runtime/run_compaction.py b/loopx/control_plane/runtime/run_compaction.py
index 2774080f8d..52df8b7ba1 100644
--- a/loopx/control_plane/runtime/run_compaction.py
+++ b/loopx/control_plane/runtime/run_compaction.py
@@ -358,9 +358,12 @@ def compact_run_base(
         compact["subagent_count"] = len(subagents)
 
     if isinstance(run.get("recommended_action_resolution"), dict):
-        resolution = run["recommended_action_resolution"]
-        if resolution.get("step_revision"):
-            compact["recommended_action_resolution"] = dict(resolution)
+        from ..work_items.recommendation_source_io import recommendation_run_decision
+
+        if recommendation_run_decision(run) is not None:
+            compact["recommended_action_resolution"] = dict(run["recommended_action_resolution"])
+            if "json_path" in run:
+                compact["json_path"] = run["json_path"]
     return compact
 
 
diff --git a/loopx/control_plane/runtime/run_context_retention.py b/loopx/control_plane/runtime/run_context_retention.py
index 16687ba82d..9496c8bce5 100644
--- a/loopx/control_plane/runtime/run_context_retention.py
+++ b/loopx/control_plane/runtime/run_context_retention.py
@@ -29,7 +29,7 @@
     "latest_evidence_delivery_run",
 )
 SEMANTIC_CONTEXT_RUN_PAYLOAD_FIELDS = {
-    "latest_lane_step_run": ("generated_at", "goal_id", "agent_id", "recommended_action_resolution"),
+    "latest_lane_step_run": ("generated_at", "json_path", "goal_id", "agent_id", "recommended_action_resolution"),
     "latest_agent_vision_run": (
         "generated_at",
         "agent_id",
@@ -135,6 +135,8 @@ def goal_semantic_history_from_runs(
     bounded list.
     """
 
+    from ..work_items.recommendation_source_io import recommendation_run_decision
+
     contexts: dict[str, dict[str, Any]] = {}
     resolved_agent_vision: set[str] = set()
     latest_owner_correction_run: dict[str, Any] | None = None
@@ -162,8 +164,8 @@ def goal_semantic_history_from_runs(
                     active_blocked_retry_runs.append(run)
         context = contexts.setdefault(agent_id, {"agent_id": agent_id})
 
-        resolution = run.get("recommended_action_resolution")
-        if "latest_lane_step_run" not in context and isinstance(resolution, dict) and resolution.get("step_revision"):
+        if ("latest_lane_step_run" not in context and run.get("agent_id") == agent_id
+                and recommendation_run_decision(run) is not None):
             context["latest_lane_step_run"] = run
         checkpoint = run.get("vision_checkpoint")
         if "latest_vision_checkpoint_run" not in context and isinstance(
diff --git a/loopx/control_plane/work_items/recommendation_source_io.py b/loopx/control_plane/work_items/recommendation_source_io.py
index 1bf457cca9..339a1b2432 100644
--- a/loopx/control_plane/work_items/recommendation_source_io.py
+++ b/loopx/control_plane/work_items/recommendation_source_io.py
@@ -64,25 +64,27 @@ def recommendation_runs(runtime_root: Path, goal_id: str) -> list[dict[str, Any]
         key=lambda item: (*chronology_key(item[1].get("generated_at")), item[0]), reverse=True)]
 
 
-def latest_bound_recommendation(runs: list[dict[str, Any]], agent_id: str) -> dict[str, Any] | None:
+def recommendation_run_decision(row: dict[str, Any]) -> dict[str, Any] | None:
     # Journal decoding only. The TS owner validates source, actor and task binding.
     from .refresh_recommendation import RECOMMENDED_ACTION_SOURCE_EXPLICIT
 
+    resolution = row.get("recommended_action_resolution")
+    if not isinstance(resolution, dict):
+        return None
+    if resolution.get("recommended_action_source") == RECOMMENDED_ACTION_SOURCE_EXPLICIT:
+        # Preserve the same committed decision identity in full and retained
+        # history. A derived observation does not retire or mint a step.
+        return {**resolution, "agent_id": row.get("agent_id"), "goal_id": row.get("goal_id"),
+            "journal_source": {key: row.get(key) for key in ("generated_at", "json_path")}}
+    return resolution if resolution.get("step_revision") else None
+
+
+def latest_bound_recommendation(runs: list[dict[str, Any]], agent_id: str) -> dict[str, Any] | None:
     for row in runs:
-        resolution = row.get("recommended_action_resolution")
-        if row.get("agent_id") == agent_id and isinstance(resolution, dict) and (
-            resolution.get("step_revision")
-            or resolution.get("recommended_action_source") == RECOMMENDED_ACTION_SOURCE_EXPLICIT
-        ):
-            # Preserve an explicit newer decision as the read barrier. Ordinary
-            # derived observations still reuse the original step receipt; they
-            # do not mint a revision or revive a step before this decision.
-            if resolution.get("recommended_action_source") == RECOMMENDED_ACTION_SOURCE_EXPLICIT:
-                # Existing journal identity facts, not a new persisted decision
-                # or step. The TS owner binds them into the read fence only.
-                return {**resolution, "agent_id": row["agent_id"], "goal_id": row.get("goal_id"),
-                    "journal_source": {key: row.get(key) for key in ("generated_at", "json_path")}}
-            return resolution
+        if row.get("agent_id") == agent_id:
+            decision = recommendation_run_decision(row)
+            if decision is not None:
+                return decision
     return None
 
 
diff --git a/tests/control_plane/test_receipt_next_action_cli.py b/tests/control_plane/test_receipt_next_action_cli.py
index e8f4f7215e..fbba4bbe51 100644
--- a/tests/control_plane/test_receipt_next_action_cli.py
+++ b/tests/control_plane/test_receipt_next_action_cli.py
@@ -15,8 +15,71 @@
 
 from loopx.cli import main as cli_main
 from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted
 from loopx.control_plane.quota.settlement import read_heartbeat_settlement
 from loopx.control_plane.todos.contract import TODO_DECISION_SCOPE_OUTCOME_SCHEMA_VERSION
+from loopx.control_plane.todos.provider_projection import project_current_canonical_todos
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_retired_step_has_one_writable_basis_across_status_and_quota(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str,
+) -> None:
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = fixture._write_fixture(tmp_path)
+    state = fixture._configure_read_only_todo(project)
+    fixture._initialize_fixture_authority(project, runtime, provider=provider)
+    project_current_canonical_todos(registry_path=registry, runtime_root=runtime, goal_id=fixture.GOAL_ID)
+    original_state = state.read_bytes()
+    authority = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=fixture.GOAL_ID)
+
+    def run(*args: str) -> dict:
+        code, payload = fixture._run_cli(registry, runtime, *args, cwd=project)
+        assert code == 0, payload
+        return payload
+
+    def current_routes() -> tuple[dict, dict]:
+        status = run("status", "--include-task-graph", "--limit", "1")
+        item = next(row for row in status["attention_queue"]["items"]
+                    if row["goal_id"] == fixture.GOAL_ID)
+        action = next(row for row in item["agent_next_actions"]
+                      if row["agent_id"] == fixture.AGENT_ID)
+        quota = run("quota", "should-run", "--goal-id", fixture.GOAL_ID,
+                    "--agent-id", fixture.AGENT_ID, "--codex-app", "--limit", "1")
+        assert quota["effective_action"] == "normal_run", quota
+        return action, quota["agent_lane_next_action"]
+
+    refresh = ("refresh-state", "--goal-id", fixture.GOAL_ID,
+               "--agent-id", fixture.AGENT_ID, "--no-global-sync", "--suppress-external-sinks")
+    initial, initial_quota = current_routes()
+    old_basis = initial["next_action_basis"]
+    assert initial_quota["next_action_basis"] == old_basis
+    run(*refresh, "--next-action", "Continue the old experiment.", "--next-action-basis", old_basis)
+    guidance = "The evidence invalidated the experiment; choose another method."
+    run(*refresh, "--recommended-action", guidance)
+    # Later observations push the explicit decision outside both recent windows.
+    for _ in range(2):
+        run(*refresh)
+    retired, retired_quota = current_routes()
+    assert "next_step" not in retired
+    assert "next_step" not in retired_quota
+    assert retired["next_action_basis"] == retired_quota["next_action_basis"] != old_basis
+    index = runtime / "goals" / fixture.GOAL_ID / "runs/index.jsonl"
+    before = index.read_bytes()
+    code, rejected = fixture._run_cli(registry, runtime, *refresh,
+        "--next-action", "Continue the old experiment.", "--next-action-basis", old_basis, cwd=project)
+    assert code == 1 and rejected["error_code"] == "next_action_basis_conflict", rejected
+    assert index.read_bytes() == before
+    run(*refresh, "--next-action", "Evaluate the newly supported route.",
+        "--next-action-basis", retired_quota["next_action_basis"])
+    run(*refresh)
+    current, current_quota = current_routes()
+    assert current["next_step"] == current_quota["next_step"] == "Evaluate the newly supported route."
+    assert current["next_action_basis"] == current_quota["next_action_basis"]
+    assert state.read_bytes() == original_state
+    after = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=fixture.GOAL_ID)
+    for key in ("provider_revision", "cursor", "todos"):
+        assert after[key] == authority[key]
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
diff --git a/tests/control_plane/test_run_context_retention.py b/tests/control_plane/test_run_context_retention.py
index d1b035cd17..781e1f7aec 100644
--- a/tests/control_plane/test_run_context_retention.py
+++ b/tests/control_plane/test_run_context_retention.py
@@ -310,6 +310,34 @@ def test_recent_runs_are_strictly_bounded() -> None:
     assert latest_runs_with_agent_context(runs, limit=0) == []
 
 
+def test_explicit_lane_decision_keeps_source_identity_beyond_recent_history() -> None:
+    from loopx.control_plane.goals.goal_frontier.semantic_history import latest_lane_recommendation_from_status
+
+    decision = _run("2026-08-09T00:01:00Z", json_path="runs/decision.json",
+        recommended_action_resolution={"schema_version": "refresh_recommendation_v0",
+            "recommended_action_source": "explicit_arg", "recommended_action": "Choose another method."})
+    runs = [
+        _run(f"2026-08-09T00:02:{second:02d}Z") for second in reversed(range(40))
+    ] + [
+        _run("2026-08-09T00:01:30Z", agent_id=OTHER_AGENT_ID,
+            recommended_action_resolution={"recommended_action_source": "explicit_arg",
+                "recommended_action": "Review the independent route."}),
+        decision,
+        _run("2026-08-09T00:00:00Z",
+            recommended_action_resolution={"step_revision": "old-step"}),
+    ]
+    history = compact_goal_semantic_history(goal_semantic_history_from_runs(runs), compact_run=compact_run)
+    assert _agent_context(history, AGENT_ID)["latest_lane_step_run"]["json_path"] == "runs/decision.json"
+    status = {"run_history": {"goals": [{"id": "semantic-goal",
+        "semantic_history": history, "latest_runs": latest_runs_with_agent_context(runs, limit=1)}]}}
+    resolution = latest_lane_recommendation_from_status(status, goal_id="semantic-goal", agent_id=AGENT_ID)
+    assert resolution["recommended_action_source"] == "explicit_arg"
+    assert "step_revision" not in resolution
+    assert resolution["journal_source"] == {
+        "generated_at": decision["generated_at"], "json_path": "runs/decision.json"}
+    assert resolution["agent_id"] == AGENT_ID and resolution["goal_id"] == "semantic-goal"
+
+
 def test_agent_lane_status_keeps_only_selected_semantic_context() -> None:
     raw_history = goal_semantic_history_from_runs(
         [