test: reconcile release regressions with shipped contracts

tests/control_plane/test_heartbeat_agent_input.py

@@ -65,7 +65,7 @@ def test_thin_cli_preserves_readable_admission_and_authority_with_host_scope(tmp
     packet = json.loads(subprocess.check_output(command, text=True))
     assert packet["ok"] is True
     budget = packet["interface_budget"]
-    assert budget["max_chars"] == 4500
+    assert budget["max_chars"] == 4900
     assert budget["budget_char_count"] <= budget["max_chars"]
     assert budget["within_budget"] is True
     body = packet["task_body"]

tests/control_plane/test_heartbeat_envelope_transport.py

@@ -82,13 +82,13 @@ def test_capture_preserves_full_dispatcher_and_envelope_needs_explicit_opt_in(tm
     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_500
-    assert captured_only['interface_budget']['max_chars'] == 4_500
+    assert ordinary['interface_budget']['max_chars'] == 4_900
+    assert captured_only['interface_budget']['max_chars'] == 4_900
     oversized = build_heartbeat_prompt(
         **no_memory, decision_output_root=tmp_path, turn_envelope=True,
-        agent_scopes=[letter * 320 for letter in 'stu'],
+        agent_scopes=[letter * 320 for letter in 'stuv'],
     )
-    assert oversized['interface_budget']['max_chars'] == 4_900
+    assert oversized['interface_budget']['max_chars'] == 5_300
     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')):
@@ -163,7 +163,7 @@ def test_bound_captured_prompt_fits_budget_without_reward_memory(
         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_900
+        assert payload['interface_budget']['max_chars'] == 5_300
         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

tests/control_plane/test_heartbeat_registry_route.py

@@ -26,7 +26,8 @@ def _route(command: str, registry: Path, runtime: Path) -> None:
 
 
 def test_budget_normalizes_shell_quoted_registry_before_goal_substrings():
-    registry = "/fixture/route-fixture's authority/" + "long-directory/" * 300 + "registry.json"
+    # The raw route must still exceed thin's current allowance before normalization.
+    registry = "/fixture/route-fixture's authority/" + "long-directory/" * 400 + "registry.json"
     body = "loopx --registry " + shlex.quote(registry) + " quota should-run --goal-id route-fixture"
     budget = build_interface_budget(
         task_body=body, goal_id="route-fixture", active_state="/fixture/state.md",

tests/control_plane/test_quota_settlement_cli.py

@@ -2057,7 +2057,7 @@ def test_receipt_retry_keeps_open_vision_quiet_through_real_cli(
     if human_item == "user_action":
         assert waiting["interaction_contract"]["user_channel"]["non_gating_notice"] is True
         assert "response_plan" not in waiting["interaction_contract"]
-        assert waiting["scheduler_hint"]["reason_code"] == "blocked_wait"
+        assert waiting["scheduler_hint"]["reason_code"] == "interaction_agent_scope_wait"
         assert waiting["scheduler_hint"]["action"] != "backoff_waiting_for_user"
         assert waiting["scheduler_hint"]["pending_retry"]["due_at"] == closed["blocked_retry"]["due_at"]
     envelope = build_turn_envelope(waiting)

tests/control_plane/test_replan_successor_durable_ack.py

@@ -9,6 +9,7 @@
 import pytest
 
 from loopx.cli import main as cli_main
+from loopx.control_plane.effect_runtime import EffectRuntimeRejected
 from loopx.control_plane.status.autonomous_replan_projection import (
     AUTONOMOUS_REPLAN_PERIODIC_RUN_THRESHOLD,
     autonomous_replan_obligation_from_runs,
@@ -107,10 +108,18 @@ def test_cli_completed_todo_successor_refresh_resets_periodic_window(tmp_path: P
 def test_transition_cannot_settle_a_different_turn_guard(guard) -> None:
     runs = history()
     obligation = autonomous_replan_obligation_from_runs(runs, agent_todos={}, agent_id=AGENT)
-    assert enforce_open_replan_writeback(newest_first_runs=runs,
+    kwargs = dict(newest_first_runs=runs,
         state_text=successor_state(obligation["obligation_id"]), agent_id=AGENT,
         goal_id=GOAL, guard_scoped=True,
-        guard_semantic_replan_obligation_id=guard) is None
+        guard_semantic_replan_obligation_id=guard)
+    if guard is None:
+        assert enforce_open_replan_writeback(**kwargs) is None
+    else:
+        # An explicitly selected identity must reconstruct its receipt-bound
+        # source; it cannot silently fall through to an unselected obligation.
+        with pytest.raises(EffectRuntimeRejected) as rejected:
+            enforce_open_replan_writeback(**kwargs)
+        assert rejected.value.diagnostic_code == "heartbeat_receipt_identity_conflict"
 
 
 def test_transition_settles_only_exact_guard_and_owner() -> None:

tests/control_plane/test_scheduler_arbitration_transport.py

@@ -160,7 +160,9 @@ def observe(method, params, **kwargs):
         assert error.value.diagnostic_code == "runtime_request_failed"
         assert isinstance(error.value.__cause__, RecursionError)
     assert modes == [True]
-    assert snapshot_directories == []
+    # The shared transport may allocate its response sink before encoding.
+    # Rejected input must still remove every private directory and file.
+    assert all(not directory.exists() for directory in snapshot_directories)
     assert_private_files_removed(Path(tempfile.gettempdir()))
     assert effect_runtime.effect_runtime_result("runtime.ping", {})["pid"] == pid