Preserve producer-bound heartbeat Turn identities

loopx/control_plane/heartbeat/task_body.py

@@ -64,6 +64,11 @@ def bind_exact_turn_settlement_task_body(
             placeholder,
             render_turn_bootstrap(value=turn_instance_id),
         )
+    if bound != task_body:
+        # The producer already supplied the timestamp. Its concrete assignment
+        # and separate guard still carry the original identity on retries;
+        # only reusable wake templates need the substitution instruction.
+        bound = bound.replace(f"{HEARTBEAT_TURN_BOOTSTRAP_RULE}\n", "", 1)
     plan_path = ("writeback.settlement_plan.ordered_steps" if captured_envelope
                  else "interaction_contract.cli_channel.settlement_plan.ordered_steps")
     return (

tests/control_plane/test_heartbeat_agent_input.py

@@ -158,6 +158,34 @@ def test_thin_agent_input_keeps_exact_turn_and_bootstrap_identity_when_present()
     assert projected["bootstrap"] is True
 
 
+@pytest.mark.parametrize("mode", ["full", "compact", "brief", "thin"])
+@pytest.mark.parametrize("windows", [False, True])
+def test_bound_prompt_keeps_original_turn_without_wake_time_substitution(mode, windows, monkeypatch):
+    import loopx.project_prompt as prompt_shell
+
+    monkeypatch.setattr(prompt_shell, "_is_windows_host", lambda: windows)
+    args = dict(goal_id="bound-heartbeat", agent_id="agent-a",
+                runtime_profile="codex_app_heartbeat", **{mode: True})
+    dynamic = build_heartbeat_prompt(**args)
+    bound = build_heartbeat_prompt(**args, turn_instance_id="original-turn")
+    # A reusable wake template still needs timestamp substitution. A producer-
+    # bound prompt must preserve that original identity, including on retry.
+    assert "Per wake, replace `<current_time_iso>` once." in dynamic["task_body"]
+    assert "reuse the value on retries" in dynamic["task_body"]
+    assert "Per wake, replace" not in bound["task_body"]
+    assert "<current_time_iso>" not in bound["task_body"]
+    assignment = ('$LOOPX_TURN = "original-turn"' if windows
+                  else "LOOPX_TURN=original-turn")
+    lines = bound["task_body"].splitlines()
+    assert lines.count(assignment) == 1
+    assert bound["quota_guard_command"] in lines
+    assert lines.index(assignment) < lines.index(bound["quota_guard_command"])
+    # Binding does not replace the dynamic shell read with an unscoped guard.
+    assert "--turn-instance-id" in bound["quota_guard_command"]
+    assert "LOOPX_TURN" in bound["quota_guard_command"]
+    assert "interaction_contract.cli_channel.settlement_plan.ordered_steps" in bound["task_body"]
+
+
 def test_thin_agent_input_error_is_actionable_without_generator_diagnostics() -> None:
     generated = build_heartbeat_prompt_error_payload(
         goal_id="heartbeat-agent-input",

tests/control_plane/test_heartbeat_envelope_transport.py

@@ -354,7 +354,7 @@ def complete_page(value):
         '--todo-id', ALTERNATIVE_TODO_ID, '--turn-instance-id', TURN_ID,
         '--classification', 'validated_progress', '--delivery-batch-scale', 'implementation',
         '--delivery-outcome', 'outcome_progress', '--delivery-boundary', 'in_flight_continuation',
-        '--no-global-sync', '--suppress-external-sinks')
+        '--no-global-sync', '--suppress-external-sinks', cwd=project)
     assert rc == 0, refresh
     spend = refresh['settlement_owed']['command']
     for _ in range(2):