fix(heartbeat): qualify known work on the first current guard

docs/heartbeat-automation-prompt.md

@@ -689,6 +689,19 @@ For every automatic heartbeat turn, the agent-facing checklist is:
 
 1. Guard first: `quota should-run --turn-instance-id <HEARTBEAT_TURN_ID>` using
    this trigger's `<current_time_iso>` and reusing it for same-heartbeat retries.
+   If the caller has already chosen the unfinished work for this new Turn,
+   append `--todo-id <chosen-id>` to that first guard. This requests current
+   qualification and complete required context in one call; it does not adopt
+   the previous Turn's authority, read coverage or settlement. Read the current
+   contract before work and follow rejection/recovery when the choice is stale
+   or gated. When the choice is unknown, discover candidates and then use the
+   returned `selection_command`. Never guess a Todo or change an existing Turn
+   binding to avoid a read. Same-Turn compaction alone does not start this flow.
+   若本次新 Turn 已明确继续哪项未完成工作，可在首次 guard 中直接附加
+   `--todo-id <chosen-id>`，一次取得当前资格及必要上下文。旧任务引用不继承旧
+   权限、阅读覆盖或结算身份；仍须读取当前合同，过期或受限时按返回结果恢复。
+   尚未选定工作则先发现候选，再用 `selection_command`；不猜测任务、不为少读
+   一次而更换既有 Turn 绑定，同一 Turn 的压缩恢复也不触发这条新准入路径。
    If `loopx` is not initially on PATH, export `$HOME/.local/bin:$PATH`
    and run the local installer fallback before declaring preflight failure.
 2. If `should_run=false` with `state=operator_gate`, ask the user/controller the

loopx/control_plane/heartbeat/rules.py

@@ -101,8 +101,11 @@
     "Per wake, replace `<current_time_iso>` once. Run assignment and guard as separate "
     "statements in one shell, not a command-prefix assignment; reuse the value on retries."
 )
+HEARTBEAT_KNOWN_CHOICE_RULE = (
+    "Known new-Turn choice: add `--todo-id <id>` to first guard; otherwise "
+)
 HOST_LOOP_QUOTA_DISPATCH_RULE = (
-    "Quota: use selection_command when required; "
+    HEARTBEAT_KNOWN_CHOICE_RULE + "use selection_command when required; "
     "re-enter as instructed, do/verify authorized work. "
     "Settle with current settlement_plan.ordered_steps, else next_cli_actions; "
     "preserve exact identities/effect ids and actual outcomes. Never reconstruct "

loopx/control_plane/heartbeat/task_body.py

@@ -11,6 +11,7 @@
     DEFAULT_PERMISSION_RULE,
     HEARTBEAT_NOTIFICATION_RULE_SHORT,
     HEARTBEAT_TURN_BOOTSTRAP_RULE,
+    HEARTBEAT_KNOWN_CHOICE_RULE,
     HEARTBEAT_VISION_WRITEBACK_RULE_SHORT,
     HOST_LOOP_QUOTA_DISPATCH_RULE,
     HOST_LOOP_SAFETY_RULE,
@@ -680,7 +681,7 @@ def render_thin_heartbeat_task_body(
 When present, read inline `agent_context` before planning. Read `detail_ref.full_decision` for
 selection, replan, missing/truncated commands or an `agent_context.detail_ref`:
 same-invocation observation, not fresh admission.
-Selection uses the saved `interaction_contract.cli_channel.selection_command`.
+{HEARTBEAT_KNOWN_CHOICE_RULE}use the saved `interaction_contract.cli_channel.selection_command`.
 Keep all supplied guard transport flags (--turn-envelope, --decision-output-root,
 and --decision-handoff-socket when present) on every fresh selection/repair/replan
 guard; each call saves its own immutable observation.

tests/control_plane/test_scoped_gate_explicit_choice.py

@@ -70,6 +70,48 @@ def test_default_fallback_is_unchanged():
     assert packet["selected_todo"]["todo_id"] == "todo_recommended"
 
 
+@pytest.mark.parametrize("provider", ["sqlite", "file"])
+@pytest.mark.parametrize("choice_exists", [True, False])
+def test_generated_heartbeat_known_choice_uses_one_current_guard(tmp_path, provider, choice_exists):
+    from loopx.heartbeat_prompt import build_heartbeat_prompt
+    from tests.control_plane.test_quota_settlement_cli import (
+        AGENT_ID, ALTERNATIVE_TODO_ID, GOAL_ID, TURN_ID,
+        _configure_selectable_alternative, _heartbeat_receipt_count,
+        _initialize_fixture_authority, _run_generated_cli, _spend_run_count, _write_fixture,
+    )
+
+    project, runtime, registry = _write_fixture(tmp_path)
+    _configure_selectable_alternative(project)
+    _initialize_fixture_authority(project, runtime, provider=provider)
+    prompt = build_heartbeat_prompt(
+        goal_id=GOAL_ID, agent_id=AGENT_ID, registered_agents=[AGENT_ID],
+        runtime_profile="codex_app_heartbeat", turn_instance_id=TURN_ID,
+        registry_path=registry, runtime_root=runtime, thin=True,
+    )
+    body = prompt["task_body"]
+    for obligation in ("Known new-Turn choice", "--todo-id <id>", "first guard",
+                       "identities/effect ids", "selection_command"):
+        assert obligation in body
+    command = prompt["quota_guard_command"].replace('"${LOOPX_TURN:?}"', shlex.quote(TURN_ID))
+    assert "--todo-id" not in command  # Generation must not silently choose work.
+    chosen = ALTERNATIVE_TODO_ID if choice_exists else "todo_missing"
+    rc, result = _run_generated_cli(command + " --todo-id " + chosen,
+                                    registry_path=registry, cwd=project)
+    if choice_exists:
+        assert rc == 0, json.dumps(result)
+        assert result["selected_todo"]["todo_id"] == chosen
+        channel = result["interaction_contract"]
+        assert channel["agent_channel"]["delivery_allowed"]
+        assert channel["agent_channel"]["work_context"]["complete"]
+        assert not channel["cli_channel"].get("selection_required", False)
+        assert _heartbeat_receipt_count(runtime, TURN_ID) == 1
+    else:
+        assert rc != 0, result
+        assert not result["interaction_contract"]["agent_channel"]["delivery_allowed"], json.dumps(result)
+        assert _heartbeat_receipt_count(runtime, TURN_ID) == 0
+    assert _spend_run_count(runtime) == 0
+
+
 @pytest.mark.parametrize("provider", ["sqlite", "file"])
 def test_cli_begin_selects_once_and_replay_keeps_original_identity(tmp_path, provider):
     from tests.control_plane.test_quota_settlement_cli import (