fix(heartbeat): qualify inline scheduler prompts without memory

docs/reference/contracts/interface-budget-contract.md

@@ -409,6 +409,52 @@ three affected JSON surfaces; dense-to-dense changes retain ordinary limits.
 
 ### Native fixture qualification
 
+The opt-in captured TurnEnvelope heartbeat has a separate 400-character body
+allowance above ordinary thin: 4,600 without Reward Memory. It retains complete
+runtime/capture routes, same-invocation recovery, inline ACK conditions and the
+outcome guidance. Ordinary thin and capture-only thin remain at 4,200; the
+optional Reward Memory allowance remains 640. Native Goal's 4,000-character hard
+limit, other mode limits, JSON/key/line budgets and execution authority do not change.
+
+The original no-memory review input at `e552168e3` / `9197abd8d` measured
+3,551 / 3,653 against 3,600, a real historical failure. The independently
+reproduced native CLI workload below binds Goal/Agent/Turn, registry and runtime,
+uses a 128-character absolute root and an existing capture directory, and keeps
+all paths in the output. File and SQLite give the same measurements:
+
+| No-memory captured thin body | Default scope | 320-character scope |
+| --- | --- | --- |
+| Original base `e552168e3`, ceiling 3,600 | 3,587, pass | 3,887, overflow |
+| Original head `9197abd8d`, ceiling 3,600 | 3,689, overflow | 3,989, overflow |
+| Target main `82b1a8c72`, ceiling 4,200 | 4,110, pass | 4,410, overflow |
+| Inline ACK + complete current guidance, ceiling 4,600 | 4,212, pass | 4,512, pass |
+
+The 102-character ACK instruction delta survives the newer outcome rollout;
+simply adopting main's 4,200 ceiling still leaves the default bound input 12
+characters over. Keeping that ceiling fails this workload; no lossless removal
+of the actor, conditions, same-capture provenance or existing outcome obligations
+was established. The bounded opt-in allowance preserves those useful clauses
+with 88 characters of scoped headroom. Longer routes/scopes still report
+overflow; this is not an arbitrary-input guarantee. Compact Agent JSON is
+4,719 / 5,019 characters (381 scoped headroom under the unchanged 5,400
+envelope ceiling); emitted stdout including its newline is 4,720 / 5,020
+characters and 4,724 / 5,024 UTF-8 bytes. Both bodies have 38 lines and both
+payloads have seven top-level keys. The ordinary emitted thin matrix keeps its separate 4,500
+ceiling. Characters, UTF-8 bytes and model tokens are distinct; these measurements
+do not establish model compliance or total-cost improvement.
+
+可选 captured TurnEnvelope 的 thin 正文单独增加 400 字符展示余量，无 Reward
+Memory 时为 4,600；普通和仅 capture 的 thin 仍为 4,200，可选 memory 余量仍为
+640。完整路径、同次捕获、ACK 条件和成果要求均保留。原评审 3,551／3,653 对
+3,600 的失败保留；上表独立真实 CLI 在相同 128 字符根路径、默认／320 字符范围下
+复现原回归，并区分新 main 的既有超限。仅同步 main 仍使默认绑定输入超出 12 字符；
+未证明可无损删除必要条款，因此仅增加这项可选展示余量，范围输入余量为 88。
+紧凑 Agent JSON 为 4,719／5,019，沿用 5,400 上限，后者余量 381；含末尾换行的
+实际 stdout 为 4,720／5,020 字符、4,724／5,024 UTF-8 字节；正文均为 38 行，
+payload 均为七个顶层键。普通 thin
+输出矩阵仍受独立的 4,500 上限约束。更长输入仍如实告警；native Goal 硬限制、
+其他模式、结构／行数、执行额度和权限均不变，也不宣称模型遵从或完整成本收益。
+
 The current outcome rollout also affects the complete thin body and emitted
 JSON, measured independently. On the same public CLI fixture at main
 `4f6e20a13`, normalized body characters are 3,306 before / 3,829 after; the

loopx/control_plane/heartbeat/budget.py

@@ -23,6 +23,11 @@
 }
 NATIVE_GOAL_HOST_MAX_CHARS = INTERFACE_BUDGET_CHARS["visible_goal"]
 
+# Opt-in captured envelopes retain transport routes and same-capture recovery.
+# The no-memory, 128-char-root/320-char-scope CLI body measures 4,512 chars.
+# Keep its 4,600-char presentation allowance separate from ordinary thin.
+CAPTURED_ENVELOPE_PROMPT_HEADROOM_CHARS = 400
+
 # 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.
@@ -72,6 +77,7 @@ def build_interface_budget(
     brief: bool = False,
     thin: bool = False,
     native_goal_host: bool = False,
+    captured_envelope: bool = False,
 ) -> dict[str, Any]:
     mode = (
         "visible_goal"
@@ -87,7 +93,16 @@ def build_interface_budget(
         if mode != "visible_goal" and _REWARD_MEMORY_PROMPT_MARKER in task_body
         else 0
     )
-    max_chars = INTERFACE_BUDGET_CHARS[mode] + reward_memory_headroom
+    captured_envelope_headroom = (
+        CAPTURED_ENVELOPE_PROMPT_HEADROOM_CHARS
+        if mode == "thin" and captured_envelope
+        else 0
+    )
+    max_chars = (
+        INTERFACE_BUDGET_CHARS[mode]
+        + reward_memory_headroom
+        + captured_envelope_headroom
+    )
     return {
         "mode": mode,
         "char_count": len(task_body),

loopx/control_plane/heartbeat/builder.py

@@ -589,6 +589,7 @@ def build_heartbeat_prompt(
             brief=brief,
             thin=thin,
             native_goal_host=native_goal_host,
+            captured_envelope=turn_envelope,
         ),
         "task_body": task_body,
     }

loopx/control_plane/heartbeat/task_body.py

@@ -643,7 +643,8 @@ def render_thin_heartbeat_task_body(
 ```
 {SCOPE_BOUNDED_WORK_RULE}
 Observed capabilities -> --available-capability, never a user gate.
-Apply `scheduler` as projected; read the saved full decision for exact ACK args.
+Apply `scheduler` as projected using inline ACK argv. Read saved scheduler detail
+only for a needed missing/truncated/referenced command; failure argv only after host update failure.
 Exact monitor settlement=no refresh/spend; auxiliary poll continues work.
 Validate work; execute `writeback.settlement_plan` and `writeback.next_cli_actions` with
 their exact identity. Do not guess commands from abbreviated previews.

tests/control_plane/test_heartbeat_envelope_transport.py

@@ -3,6 +3,7 @@
 import os
 import shlex
 import socket
+import subprocess
 import sys
 import tempfile
 import threading
@@ -13,9 +14,11 @@
 from benchmark.runtime.codex import Execution
 from benchmark.runtime.worker import heartbeat_body
 from loopx.heartbeat_prompt import build_heartbeat_prompt
+from tests.control_plane.cli_output_fixture import stable_budget_fixture_root
 from tests.control_plane.test_quota_settlement_cli import (
-    AGENT_ID, GOAL_ID, TURN_ID, ALTERNATIVE_TODO_ID, REPO_ROOT,
+    AGENT_ID, GOAL_ID, TODO_ID, TURN_ID, ALTERNATIVE_TODO_ID, REPO_ROOT,
     _write_fixture, _configure_selectable_alternative, _run_cli,
+    _initialize_fixture_authority,
     _run_generated_cli, _heartbeat_receipt_count, _spend_run_count,
 )
 
@@ -38,8 +41,24 @@ def test_capture_preserves_full_dispatcher_and_envelope_needs_explicit_opt_in(tm
     assert 'writeback.settlement_plan.ordered_steps' in payload['task_body']
     assert 'When present, read inline `agent_context` before planning.' in payload['task_body']
     assert 'an `agent_context.detail_ref`' in payload['task_body']
+    assert 'using inline ACK argv' in payload['task_body']
+    assert 'needed missing/truncated/referenced command' in payload['task_body']
+    assert 'failure argv only after host update failure' in payload['task_body']
     assert '--decision-output-root' in payload['thin_prompt_command']
     assert '--turn-envelope' in payload['thin_prompt_command']
+    # The captured allowance must neither leak into ordinary thin nor disable
+    # overflow diagnostics, including when Reward Memory is absent.
+    no_memory = dict(args, reward_memory_enabled=False)
+    ordinary = build_heartbeat_prompt(**no_memory)
+    captured_only = build_heartbeat_prompt(**no_memory, decision_output_root=tmp_path)
+    assert ordinary['interface_budget']['max_chars'] == 4_200
+    assert captured_only['interface_budget']['max_chars'] == 4_200
+    oversized = build_heartbeat_prompt(
+        **no_memory, decision_output_root=tmp_path, turn_envelope=True,
+        agent_scopes=[letter * 320 for letter in 'stu'],
+    )
+    assert oversized['interface_budget']['max_chars'] == 4_600
+    assert oversized['interface_budget']['within_budget'] is False
     for overrides in (dict(turn_instance_id=None), dict(thin=False, full=True),
                       dict(runtime_profile='codex_app_ssh_goal')):
         with pytest.raises(ValueError):
@@ -48,6 +67,81 @@ def test_capture_preserves_full_dispatcher_and_envelope_needs_explicit_opt_in(tm
         build_heartbeat_prompt(**args, decision_output_root=tmp_path / 'missing')
 
 
+@pytest.mark.parametrize('provider', ['file', 'sqlite'])
+def test_captured_scheduler_ack_is_inline_and_failure_detail_stays_on_original_source(
+    tmp_path, monkeypatch, provider,
+):
+    from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_fixture(tmp_path)
+    _initialize_fixture_authority(project, runtime, provider=provider)
+    capture = tmp_path / 'capture'
+    code, envelope = _run_cli(registry, runtime, 'quota', 'should-run',
+        '--goal-id', GOAL_ID, '--agent-id', AGENT_ID, '--todo-id', TODO_ID,
+        '--turn-instance-id', TURN_ID, '--codex-app', '--turn-envelope',
+        '--decision-output-dir', str(capture), cwd=project)
+    assert code == 0, envelope
+    original = (capture / 'decision.json').read_bytes()
+    source = json.loads(original)['scheduler_hint']['app_automation']
+    projected = envelope['scheduler']['app_automation']
+    assert projected['ack_cli_args'] == source['ack_hint']['cli_args']
+    assert 'ack_cli_args_detail_ref' not in projected
+    assert projected['failure_cli_args_detail_ref']['detail_ref'] == (
+        'full_decision.scheduler_hint.app_automation.failure_hint.cli_args'
+    )
+    before = _heartbeat_receipt_count(runtime, TURN_ID)
+    read = subprocess.run(shlex.split(envelope['detail_ref']['full_decision']),
+        capture_output=True, check=True).stdout
+    assert read == original
+    assert _heartbeat_receipt_count(runtime, TURN_ID) == before
+
+
+@pytest.mark.parametrize('provider', ['file', 'sqlite'])
+@pytest.mark.parametrize('scope_chars', [0, 320])
+def test_bound_captured_prompt_fits_budget_without_reward_memory(
+    tmp_path, monkeypatch, provider, scope_chars,
+):
+    from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    # Match the public budget workload independently of pytest/runner path length.
+    # Both runtime and capture routes remain present and count toward the body.
+    with stable_budget_fixture_root(tmp_path) as base:
+        root = (base / ('p' * (128 - len(str(base.resolve())) - 1))).resolve()
+        assert len(str(root)) == 128
+        project, runtime, registry = _write_fixture(root)
+        _initialize_fixture_authority(project, runtime, provider=provider)
+        capture = root / 'capture'
+        capture.mkdir()
+        argv = [
+            sys.executable, '-m', 'loopx.entrypoint', '--format', 'json',
+            '--registry', str(registry), '--runtime-root', str(runtime),
+            'heartbeat-prompt', '--thin', '--turn-envelope',
+            '--decision-output-root', str(capture), '--runtime-profile', 'generic_cli',
+            '--goal-id', GOAL_ID, '--agent-id', AGENT_ID, '--turn-instance-id', TURN_ID,
+        ]
+        if scope_chars:
+            argv.extend(['--agent-scope', 's' * scope_chars])
+        result = subprocess.run(
+            argv, cwd=project, capture_output=True, text=True, check=True,
+            env={**os.environ, 'PYTHONPATH': str(REPO_ROOT)},
+        )
+        payload = json.loads(result.stdout)
+        body = payload['task_body']
+        assert '--reward-memory-reflection-json' not in body
+        assert str(runtime) in body and str(capture) in body
+        assert payload['turn_instance_id'] == TURN_ID
+        assert payload['interface_budget']['max_chars'] == 4_600
+        assert payload['interface_budget']['within_budget'], payload['interface_budget']
+        assert len(result.stdout) <= 5_400  # Shipped Agent JSON, not generator metadata.
+        assert 'using inline ACK argv' in body
+        assert 'needed missing/truncated/referenced command' in body
+        assert 'failure argv only after host update failure' in body
+        assert 'same-invocation observation, not fresh admission' in body
+        assert 'goal_acceptance' in body  # Retain the complete outcome guidance too.
+
+
 @pytest.mark.parametrize('mode', ['plain', 'native-goal', 'loopx-goal', 'turn'])
 def test_incompatible_worker_rejected(mode):
     with pytest.raises(ValueError, match='turn_envelope'):
@@ -107,7 +201,6 @@ def test_real_worker_generated_guard_selection_reentry_and_settle_once(tmp_path,
     guard = next(line for line in block.splitlines() if 'quota should-run' in line)
     guard = guard.replace('${LOOPX_TURN:?}', TURN_ID)
     argv = shlex.split(guard)
-    import subprocess
     result = subprocess.run(argv, env=env, cwd=project, capture_output=True, text=True)
     first = json.loads(result.stdout)
     assert result.returncode == 0, first