fix(todo): retire old next steps after explicit guidance

docs/project-agent-todo-contract.md

@@ -591,6 +591,27 @@ 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.
+
+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.
+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
 current task. See [Next Action writeback](quota-allocation.md) for attribution,

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)

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
 
 

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(

loopx/control_plane/work_items/recommendation_source_io.py

@@ -64,12 +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"):
-            return resolution
+        if row.get("agent_id") == agent_id:
+            decision = recommendation_run_decision(row)
+            if decision is not None:
+                return decision
     return None
 
 

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);

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
@@ -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)
@@ -329,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]

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"])

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(
         [