fix(heartbeat): reserve readable Reward Memory prompt headroom

docs/heartbeat-automation-prompt.md

@@ -26,6 +26,15 @@ not just a diagnostic label. Under `DONT_NOTIFY`, repair stays internal; it
 does not grant notification authority. Keep these semantics in brief and thin
 prompts as well as the expanded contract, within their existing size budgets.
 
+Reward Memory outcome guidance is a deliberate prompt increment, not a reason
+to remove the validator, digest/evidence, writeback/readback, zero-provider-call,
+or private-material boundaries. When automatic outcome ingestion is enabled, the
+generated heartbeat body receives a fixed `+640`-character headroom (and the
+corresponding bounded line/UTF-8 allowance) for this contract. The CLI output
+differential also grants that allowance once during the none-to-v1 migration;
+unrelated later growth still uses the ordinary gate. Keep the explanation
+readable, while visible Goal prompts and other surfaces retain their own limits.
+
 Do not paste the full lifecycle protocol into the visible goal text, and do not
 use a short goal text such as "advance TODO" as the recurring automation body.
 The short text names the goal; the generated task body enforces quota, gates,

loopx/cli_commands/support_control.py

@@ -17,6 +17,7 @@
     serve_chat,
 )
 from ..control_plane.scheduler.execution_context import SchedulerRuntimeProfile
+from ..control_plane.reward_memory import reward_memory_goal_policy
 from ..dashboard_launcher import launch_dashboard, replace_existing_loopx_chat
 from ..execution_profile import execution_profile_turn_granularity
 from ..heartbeat_prequota import (
@@ -540,6 +541,13 @@ def handle_support_control_command(
                 if isinstance(registry_goal, dict)
                 else None
             )
+            reward_memory_policy = reward_memory_goal_policy(
+                registry_goal if isinstance(registry_goal, dict) else {}
+            )
+            reward_memory_enabled = bool(
+                reward_memory_policy["enabled"]
+                and reward_memory_policy["automation"].get("automatic_ingest") is True
+            )
             agent_profile = None
             if args.agent_id:
                 effective_agent_id = require_registered_agent_id(
@@ -600,6 +608,7 @@ def handle_support_control_command(
                 visible_goal_host=args.visible_goal_host,
                 turn_granularity=turn_granularity,
                 turn_instance_id=args.turn_instance_id,
+                reward_memory_enabled=reward_memory_enabled,
             )
             if args.bootstrap and payload.get("ok"):
                 from ..control_plane.heartbeat.bootstrap_prompt import goal_bootstrap

loopx/control_plane/heartbeat/budget.py

@@ -14,6 +14,12 @@
 }
 NATIVE_GOAL_HOST_MAX_CHARS = INTERFACE_BUDGET_CHARS["visible_goal"]
 
+# Reward Memory's outcome contract is intentionally readable rather than
+# squeezed into the ordinary heartbeat allowance.  Keep this fixed and tied to
+# the generated contract marker so feature-off prompts retain the old budget.
+REWARD_MEMORY_OUTCOME_PROMPT_HEADROOM_CHARS = 640
+_REWARD_MEMORY_PROMPT_MARKER = "--reward-memory-reflection-json"
+
 
 def heartbeat_prompt_mode(
     *,
@@ -55,12 +61,18 @@ def build_interface_budget(
     )
     budget_text = prompt_budget_text(task_body, goal_id=goal_id, active_state=active_state)
     budget_chars = len(budget_text)
-    max_chars = INTERFACE_BUDGET_CHARS[mode]
+    reward_memory_headroom = (
+        REWARD_MEMORY_OUTCOME_PROMPT_HEADROOM_CHARS
+        if mode != "visible_goal" and _REWARD_MEMORY_PROMPT_MARKER in task_body
+        else 0
+    )
+    max_chars = INTERFACE_BUDGET_CHARS[mode] + reward_memory_headroom
     return {
         "mode": mode,
         "char_count": len(task_body),
         "line_count": len(task_body.splitlines()),
         "budget_char_count": budget_chars,
         "max_chars": max_chars,
+        "reward_memory_headroom_chars": reward_memory_headroom,
         "within_budget": budget_chars <= max_chars,
     }

loopx/control_plane/heartbeat/builder.py

@@ -45,6 +45,8 @@
 from .rules import (
     DEFAULT_MATERIAL_QUEUE_RULE,
     DEFAULT_PERMISSION_RULE,
+    REWARD_MEMORY_OUTCOME_COMPACT_RULE,
+    REWARD_MEMORY_OUTCOME_RULE,
 )
 from .task_body import (
     bind_exact_turn_settlement_task_body,
@@ -107,6 +109,26 @@ def _select_task_body_renderer(
     return render_heartbeat_task_body
 
 
+def _reward_memory_rule_kwargs(
+    *,
+    full: bool,
+    reward_memory_enabled: bool,
+    native_goal_host: bool,
+    ark_managed_agent_goal: bool,
+) -> dict[str, str]:
+    if native_goal_host or ark_managed_agent_goal:
+        return {}
+    if not reward_memory_enabled:
+        return {"reward_memory_rule": ""}
+    return {
+        "reward_memory_rule": (
+            REWARD_MEMORY_OUTCOME_RULE
+            if full
+            else REWARD_MEMORY_OUTCOME_COMPACT_RULE
+        )
+    }
+
+
 def _heartbeat_regeneration_commands(
     *,
     cli_bin: str,
@@ -266,6 +288,7 @@ def build_heartbeat_prompt(
     visible_goal_host: str | None = None,
     turn_granularity: str | None = None,
     turn_instance_id: str | None = None,
+    reward_memory_enabled: bool = True,
 ) -> dict[str, Any]:
     if not (full or compact or brief or thin):
         thin = True
@@ -414,6 +437,12 @@ def build_heartbeat_prompt(
         brief=brief,
         compact=compact,
     )
+    reward_memory_rule_kwargs = _reward_memory_rule_kwargs(
+        full=full,
+        reward_memory_enabled=reward_memory_enabled,
+        native_goal_host=native_goal_host,
+        ark_managed_agent_goal=ark_managed_agent_goal,
+    )
     task_body = task_body_renderer(
         goal_id=goal_id,
         active_state=active_state_text,
@@ -433,6 +462,7 @@ def build_heartbeat_prompt(
         compact_prompt_command=str(commands["compact_prompt_command"]),
         brief_prompt_command=str(commands["brief_prompt_command"]),
         thin_prompt_command=str(commands["thin_prompt_command"]),
+        **reward_memory_rule_kwargs,
     )
     task_body = bind_exact_turn_settlement_task_body(
         task_body,
@@ -472,6 +502,7 @@ def build_heartbeat_prompt(
         "agent_profile": agent_profile_prompt_projection(agent_profile),
         "registered_agents": normalized_registered_agents,
         "runtime_profile": runtime_profile,
+        "reward_memory_enabled": reward_memory_enabled,
         "scheduler_execution_context": scheduler_execution_context,
         **(
             {"visible_goal_host": visible_goal_host}

loopx/control_plane/heartbeat/rules.py

@@ -34,8 +34,10 @@
     "raw/private material."
 )
 REWARD_MEMORY_OUTCOME_COMPACT_RULE = (
-    "`--reward-memory-reflection-json`: Todo validator exact digest/evidence; "
-    "else zero provider calls; no raw/private."
+    "Auto-ingest Todo: add `--reward-memory-reflection-json <reflection JSON>` "
+    "to refresh. Private stage; provider write needs Todo validator exact "
+    "digest/evidence attestation + refresh/spend readback. Else awaiting/zero "
+    "provider calls; no raw/private content."
 )
 SCHEDULER_HINT_APPLICATION_RULE = (
     "`scheduler_hint` no-spend. host_action=pause_or_delete_current_heartbeat -> "

loopx/control_plane/heartbeat/task_body.py

@@ -83,6 +83,7 @@ def render_heartbeat_task_body(
     compact_prompt_command: str,
     brief_prompt_command: str,
     thin_prompt_command: str,
+    reward_memory_rule: str = REWARD_MEMORY_OUTCOME_RULE,
 ) -> str:
     scope_block = f"\n{agent_scope_instruction}\n" if agent_scope_instruction else ""
     pr_review_pre_quota_block = (
@@ -182,7 +183,7 @@ def render_heartbeat_task_body(
    LoopX owns reset/progression state. It is scheduling only, not delivery
    permission.
 
-   {REWARD_MEMORY_OUTCOME_RULE}
+   {reward_memory_rule}
    Then use
    `heartbeat_recommendation`: `recommended_mode=run_first_read_only_map` means
    run its `command` as a real read-only map, then
@@ -283,6 +284,7 @@ def render_brief_heartbeat_task_body(
     compact_prompt_command: str,
     brief_prompt_command: str,
     thin_prompt_command: str,
+    reward_memory_rule: str = REWARD_MEMORY_OUTCOME_COMPACT_RULE,
 ) -> str:
     scope_block = f"\n{agent_scope_instruction}\n" if agent_scope_instruction else ""
     pr_review_pre_quota_block = (
@@ -321,7 +323,7 @@ def render_brief_heartbeat_task_body(
 `should_run=true`：读 compact、`status --limit 3`、`review-packet --handoff-only`；
 遵守 quota 权限/结果/handoff；outcome-floor recovery 推进 evidence 或写 blocker。
 {HOST_LOOP_QUOTA_DISPATCH_RULE}
-{REWARD_MEMORY_OUTCOME_COMPACT_RULE}
+{reward_memory_rule}
 交付并验证后，按当前 `interaction_contract.cli_channel.settlement_plan.ordered_steps`
 的精确 identity/effect 顺序结算；无 plan 时按当前 `next_cli_actions`，不使用旧 refresh/spend 配方。
 Todo验收非结算；外部等待须 open+monitor_changed+successor→重跑/继续，且不扣额；
@@ -352,6 +354,7 @@ def render_compact_heartbeat_task_body(
     compact_prompt_command: str,
     brief_prompt_command: str,
     thin_prompt_command: str,
+    reward_memory_rule: str = REWARD_MEMORY_OUTCOME_COMPACT_RULE,
 ) -> str:
     scope_block = f"\n{agent_scope_instruction}\n" if agent_scope_instruction else ""
     pr_review_pre_quota_block = (
@@ -394,7 +397,7 @@ def render_compact_heartbeat_task_body(
    Legacy/raw fallback is not owner/gate/stop authority. Treat
    `run_history.latest_runs` as drill-down only.
 
-{REWARD_MEMORY_OUTCOME_COMPACT_RULE}
+{reward_memory_rule}
 2. Goal-owned blocker: stop its path. Under `NOTIFY`, send a concrete Chinese
    blocker-push; under `DONT_NOTIFY`, repair internally and stay quiet.
    Dependency/sibling todos: record; continue audit.
@@ -639,6 +642,7 @@ def render_thin_heartbeat_task_body(
     compact_prompt_command: str,
     brief_prompt_command: str,
     thin_prompt_command: str,
+    reward_memory_rule: str = REWARD_MEMORY_OUTCOME_COMPACT_RULE,
 ) -> str:
     policy_tail = _render_compact_policy_tail(
         material_queue_rule=material_queue_rule,
@@ -672,7 +676,7 @@ def render_thin_heartbeat_task_body(
 
 P0 blocked: safe P1/P2; monitor quiet/no-spend.
 
-{REWARD_MEMORY_OUTCOME_COMPACT_RULE}
+{reward_memory_rule}
 
 {policy_tail}"""
 def render_heartbeat_generator_inputs_markdown(payload: dict[str, Any]) -> str:

loopx/heartbeat_prompt.py

@@ -20,6 +20,7 @@
 from .control_plane.heartbeat.budget import (
     INTERFACE_BUDGET_CHARS,
     NATIVE_GOAL_HOST_MAX_CHARS,
+    REWARD_MEMORY_OUTCOME_PROMPT_HEADROOM_CHARS,
     build_interface_budget,
     heartbeat_prompt_mode,
     prompt_budget_text,
@@ -59,6 +60,7 @@
     "HEARTBEAT_VISION_WRITEBACK_RULE_SHORT",
     "INTERFACE_BUDGET_CHARS",
     "NATIVE_GOAL_HOST_MAX_CHARS",
+    "REWARD_MEMORY_OUTCOME_PROMPT_HEADROOM_CHARS",
     "RUNTIME_CAPABILITY_PROJECTION_THIN_RULE",
     "RUNTIME_EXECUTION_ROUTING_RULE",
     "SCHEDULER_HINT_APPLICATION_RULE",

loopx/upgrade.py

@@ -14,6 +14,7 @@
 )
 from .execution_profile import execution_profile_turn_granularity
 from .heartbeat_prompt import build_heartbeat_prompt
+from .control_plane.reward_memory import reward_memory_goal_policy
 from .history import load_registry
 from .paths import DEFAULT_RUNTIME_ROOT, global_registry_path, resolve_runtime_root
 from .registry import registry_goals, resolve_state_file
@@ -716,6 +717,11 @@ def build_upgrade_plan(
             deferred.append(stage_deferred_goal_summary(goal, state_file))
             continue
         registered_agents = registered_agent_ids_for_goal(goal)
+        reward_memory_policy = reward_memory_goal_policy(goal)
+        reward_memory_enabled = bool(
+            reward_memory_policy["enabled"]
+            and reward_memory_policy["automation"].get("automatic_ingest") is True
+        )
         turn_granularity = execution_profile_turn_granularity(
             goal.get("execution_profile")
             if isinstance(goal.get("execution_profile"), dict)
@@ -752,6 +758,7 @@ def build_upgrade_plan(
                 available_capabilities=available_capabilities,
                 runtime_profile="codex_app_heartbeat",
                 turn_granularity=turn_granularity,
+                reward_memory_enabled=reward_memory_enabled,
             )
             summary = prompt_summary(prompt, mode)
             summary["agent_id"] = agent_id

tests/control_plane/test_heartbeat_prompt_support.py

@@ -11,6 +11,7 @@
     normalize_agent_scopes,
 )
 from loopx.control_plane.heartbeat.budget import (
+    REWARD_MEMORY_OUTCOME_PROMPT_HEADROOM_CHARS,
     build_interface_budget,
     heartbeat_prompt_mode,
     prompt_budget_text,
@@ -169,6 +170,37 @@ def test_public_facade_still_builds_and_renders_prompts() -> None:
     assert render_heartbeat_prompt_markdown(payload)
 
 
+@pytest.mark.parametrize(
+    ("mode", "base_budget"),
+    [("full", 12000), ("compact", 6500), ("brief", 3500), ("thin", 2500)],
+)
+def test_reward_memory_prompt_headroom_is_fixed_and_feature_scoped(
+    mode: str, base_budget: int
+) -> None:
+    enabled = build_heartbeat_prompt(
+        goal_id="reward-memory-budget-fixture",
+        runtime_profile="codex_app_heartbeat",
+        **{mode: True},
+        reward_memory_enabled=True,
+    )
+    disabled = build_heartbeat_prompt(
+        goal_id="reward-memory-budget-fixture",
+        runtime_profile="codex_app_heartbeat",
+        **{mode: True},
+        reward_memory_enabled=False,
+    )
+    assert "--reward-memory-reflection-json" in enabled["task_body"]
+    assert enabled["interface_budget"]["reward_memory_headroom_chars"] == (
+        REWARD_MEMORY_OUTCOME_PROMPT_HEADROOM_CHARS
+    )
+    assert enabled["interface_budget"]["max_chars"] == (
+        base_budget + REWARD_MEMORY_OUTCOME_PROMPT_HEADROOM_CHARS
+    )
+    assert "--reward-memory-reflection-json" not in disabled["task_body"]
+    assert disabled["interface_budget"]["reward_memory_headroom_chars"] == 0
+    assert disabled["interface_budget"]["max_chars"] == base_budget
+
+
 @pytest.mark.parametrize("mode", ["full", "compact", "brief", "thin"])
 def test_sizing_guidance_survives_prompt_compaction(mode: str) -> None:
     payload = build_heartbeat_prompt(goal_id="sizing-fixture", **{mode: True})
@@ -197,6 +229,10 @@ def test_reward_memory_outcome_gate_survives_app_prompt_compaction(mode: str) ->
     assert "exact" in body and "digest" in body and "evidence" in body
     assert "zero provider calls" in body
     assert "raw" in body and "private" in body
+    if mode != "full":
+        assert "Auto-ingest Todo" in body
+        assert "refresh/spend readback" in body
+        assert "no raw/private content" in body
     if mode != "full":
         assert payload["interface_budget"]["within_budget"] is True