fix(heartbeat): yield timed waits to the host scheduler

docs/project-agent-todo-contract.md

@@ -931,6 +931,37 @@ loopx todo update \
   --reason "<public-safe deferred rationale>"
 ```
 
+For a known future eligibility time, continue useful independent work first.
+When only the timed wait remains, save recoverable progress and defer the Todo
+using the timestamp supplied by the current contract or observed service:
+
+```bash
+loopx todo update \
+  --goal-id <goal-id> \
+  --todo-id <todo_id> \
+  --agent-id <registered-agent> \
+  --status deferred \
+  --resume-when resume_at:<timezone-aware-rfc3339-timestamp> \
+  --reason "Await the observed eligibility time"
+```
+
+The ordinary lifecycle update retires the applicable execution; no separate
+lease operation is needed. Settle any validated work already performed under
+the original Turn identity, then return control to the host scheduler. Waiting
+alone is not progress or a reason to spend a slot. Do not keep a model invocation
+alive with repeated sleeps, clock reads, or unchanged progress messages. The
+scheduler can recheck current state without waking the model while the wait is
+pending. At the due time, follow the projected resume/replan and fresh admission;
+the timestamp does not itself authorize execution, reset a service cooldown,
+extend a budget, or override an explicit stop. Unknown-time observations retain
+their projected monitor contract; do not invent a retry time or successor.
+
+已知恢复时间且没有可推进的独立工作时，保存进展，通过普通 Todo 更新设置
+`deferred` 和 `resume_at`，按原 Turn 身份完成已有工作的结算，再交还宿主调度器。
+不需要单独管理租约，也不应让模型反复睡眠、读时钟或播报无变化的进度。等待本身不
+算成果、不扣额；到期只触发生命周期恢复判断，仍须按当前准入、预算和停止条件执行。
+未知时间的外部反馈沿用已有 monitor 合同，不臆造时钟或后继任务。
+
 When an open advancement slice is waiting for a monitor observation, keep it
 visible and atomically bind both the wait and an independent runnable
 successor:

loopx/control_plane/heartbeat/budget.py

@@ -10,23 +10,23 @@
 # Keep useful admission and authority instructions readable; do not abbreviate
 # them solely to fit a historical character ceiling.
 INTERFACE_BUDGET_CHARS = {
-    # Complete outcome intake/review adds 523 authored characters. Matched
-    # no-memory bodies with the supported 320-character profile scope measure
-    # 12,688 / 7,531 / 5,285; preserve the instructions and bound their growth.
-    "full": 12_800,
-    "compact": 7_600,
-    "brief": 5_400,
-    # Matched public CLI fixture: 3,306 -> 3,829 normalized body chars.
-    # The same supported 320-char scope reaches 4,118, leaving 82 chars
-    # above the complete outcome, authority, continuation and settlement rules.
-    "thin": 4_200,
+    # Timed-wait guidance replaces unconditional keep-open advice. With the
+    # same 320-char scope, explicit registry and no reward memory, normalized
+    # base -> head sizes are 12,731 -> 12,982; 7,601 -> 7,861;
+    # 5,355 -> 5,615; 4,057 -> 4,378. Preserve evidence, independent work,
+    # original settlement and fresh admission rather than abbreviating them.
+    "full": 13_100,
+    "compact": 8_000,
+    # The same generic CLI + 320-char scope with explicit runtime also needs
+    # 5,715 chars; retain those routes instead of shortening the instructions.
+    "brief": 5_800,
+    "thin": 4_500,
     "visible_goal": 4_000,
 }
 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.
+# Keep captured-envelope transport headroom separate from ordinary thin.
 CAPTURED_ENVELOPE_PROMPT_HEADROOM_CHARS = 400
 
 # Reward Memory's outcome contract is intentionally readable rather than

loopx/control_plane/heartbeat/rules.py

@@ -111,15 +111,21 @@
     "preserve exact identities/effect ids and actual outcomes. Never reconstruct "
     "static refresh/spend; ambiguous effects need readback/recovery, not blind retry."
 )
+HOST_LOOP_WAIT_RULE = (
+    "No useful independent work before an evidenced due time: save progress; "
+    "`todo update --status deferred --resume-when resume_at:<RFC3339> --reason ...` "
+    "settle original Turn, then yield to host scheduler, not model sleep/poll loops. "
+    "Fresh admission on resume. Other waits: projected monitor/continuation. "
+    "No spend for waiting alone."
+)
 HOST_LOOP_TODO_CLOSEOUT_RULE = (
     "Todo done does not settle the Turn. Complete it under its original declaration "
     "and current lease; no invented successor or --no-follow-up. Then follow the "
-    "current settlement plan. External wait: keep open; follow projected "
-    "monitor/continuation; continue only admitted work. No wait spend."
+    "current settlement plan. Continue only admitted work. " + HOST_LOOP_WAIT_RULE
 )
 HOST_LOOP_TODO_CLOSEOUT_COMPACT_RULE = (
-    "Todo done≠Turn settled; no invented successor/--no-follow-up. Wait: keep open; "
-    "follow projected monitor/continuation; continue only admitted work. No wait spend."
+    "Todo done≠Turn settled; no invented successor/--no-follow-up. "
+    "Continue only admitted work. " + HOST_LOOP_WAIT_RULE
 )
 CODEX_NATIVE_GOAL_UNCHANGED_WAIT_RULE = (
     "\n\nNative Codex `/goal` owns blocked state. Recheck quota at the "

loopx/control_plane/heartbeat/task_body.py

@@ -723,7 +723,7 @@ def render_thin_heartbeat_task_body(
 {RUNTIME_CAPABILITY_PROJECTION_THIN_RULE}
 {SCHEDULER_HINT_THIN_RULE}
 {HEARTBEAT_VISION_WRITEBACK_RULE_SHORT}
-Todo done≠settled; no invented successor. Wait=open; projected monitor/continuation; re-enter guard.
+{HOST_LOOP_TODO_CLOSEOUT_COMPACT_RULE}
 `agent_read_required`: drain/read/triage before work; settle/ACK.
 
 P0 blocked: admitted P1/P2; no wait spend.

tests/control_plane/test_heartbeat_brief_budget.py

@@ -108,7 +108,11 @@ def test_brief_keeps_complete_routes_with_128_character_windows_root() -> None:
         "preserve exact identities/effect ids and actual outcomes",
         "Never reconstruct static refresh/spend; ambiguous effects need readback/recovery, not blind retry",
         "Todo done≠Turn settled; no invented successor/--no-follow-up",
-        "No wait spend",
+        "No useful independent work before an evidenced due time",
+        "--status deferred --resume-when resume_at:",
+        "settle original Turn, then yield to host scheduler",
+        "Fresh admission on resume",
+        "No spend for waiting alone",
         "Finish Goal only on terminal no-follow-up",
         "Destructive Git/production requires explicit authorization",
     ):

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_200
-    assert captured_only['interface_budget']['max_chars'] == 4_200
+    assert ordinary['interface_budget']['max_chars'] == 4_500
+    assert captured_only['interface_budget']['max_chars'] == 4_500
     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']['max_chars'] == 4_900
     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_600
+        assert payload['interface_budget']['max_chars'] == 4_900
         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_notification_rule.py

@@ -127,12 +127,15 @@ def test_todo_closeout_does_not_invent_successors_or_skip_turn_settlement(tmp_pa
         assert "no invented successor" in body
         assert "settlement_plan" in body
         assert "projected monitor/continuation" in body
-        assert "keep open" in body or "Wait=open" in body
-        assert "no wait spend" in body.lower()
-        assert "continue only admitted work" in body or "admitted P1/P2" in body
+        assert "No useful independent work before an evidenced due time" in body
+        assert "--status deferred --resume-when resume_at:" in body
+        assert "settle original Turn, then yield to host scheduler" in body
+        assert "Fresh admission on resume" in body
+        assert "No spend for waiting alone" in body
+        assert "continue only admitted work" in body.lower() or "admitted P1/P2" in body
         for stale in ("Done -> successor first", "Done->successor",
                       "open + monitor_changed + successor", "wait->monitor+successor/work",
-                      "no-follow-up completion"):
+                      "no-follow-up completion", "Wait=open"):
             assert stale not in body
 
 
@@ -212,9 +215,9 @@ def test_language_policy_preserves_blocker_and_native_goal_continuation() -> Non
     brief = " ".join(render_brief_heartbeat_task_body(**kwargs).split())
     assert "settlement_plan.ordered_steps" in brief
     assert "preserve exact identities/effect ids and actual outcomes" in brief
-    assert "Wait: keep open; follow projected monitor/continuation" in brief
-    assert "continue only admitted work" in brief
-    assert "No wait spend" in brief
+    assert "Other waits: projected monitor/continuation" in brief
+    assert "continue only admitted work" in brief.lower()
+    assert "No spend for waiting alone" in brief
     assert "Finish Goal only on terminal no-follow-up" in brief
 
     for renderer in (

tests/control_plane/test_heartbeat_prompt_support.py

@@ -97,8 +97,8 @@ def test_native_profile_long_routes_retain_one_complete_settlement_contract() ->
 
 
 @pytest.mark.parametrize("mode,limit", [
-    ("full", 12800), ("compact", 7600), ("brief", 5400),
-    ("thin", 4200), ("visible_goal", 4000),
+    ("full", 13100), ("compact", 8000), ("brief", 5800),
+    ("thin", 4500), ("visible_goal", 4000),
 ])
 def test_prompt_body_limit_remains_independent_of_json_envelope(mode: str, limit: int) -> None:
     for size in (limit, limit + 1):
@@ -356,19 +356,23 @@ def test_brief_diagnostics_keep_only_the_selected_prompt_command() -> None:
 
 @pytest.mark.parametrize(
     ("mode", "base_budget"),
-    [("full", 12800), ("compact", 7600), ("brief", 5400), ("thin", 4200)],
+    [("full", 13100), ("compact", 8000), ("brief", 5800), ("thin", 4500)],
 )
 @pytest.mark.parametrize("scope_chars", [0, 320])
+@pytest.mark.parametrize("runtime_profile", ["codex_app_heartbeat", "generic_cli"])
+@pytest.mark.parametrize("explicit_route", [False, True])
 def test_reward_memory_prompt_headroom_is_fixed_and_feature_scoped(
-    mode: str, base_budget: int, scope_chars: int
+    mode: str, base_budget: int, scope_chars: int, runtime_profile: str, explicit_route: bool,
 ) -> None:
     profile = {"scope_summary": "x" * scope_chars} if scope_chars else None
+    route = {"registry_path": "/fixture/registry.json", "runtime_root": "/fixture/runtime"} if explicit_route else {}
     enabled = build_heartbeat_prompt(
         goal_id="reward-memory-budget-fixture",
         agent_id="worker-a",
         registered_agents=["worker-a"],
         agent_profile=profile,
-        runtime_profile="codex_app_heartbeat",
+        runtime_profile=runtime_profile,
+        **route,
         **{mode: True},
         reward_memory_enabled=True,
     )
@@ -377,7 +381,8 @@ def test_reward_memory_prompt_headroom_is_fixed_and_feature_scoped(
         agent_id="worker-a",
         registered_agents=["worker-a"],
         agent_profile=profile,
-        runtime_profile="codex_app_heartbeat",
+        runtime_profile=runtime_profile,
+        **route,
         **{mode: True},
         reward_memory_enabled=False,
     )
@@ -408,6 +413,28 @@ def test_sizing_guidance_survives_prompt_compaction(mode: str) -> None:
     assert "--fine-grained" not in payload["quota_guard_command"]
 
 
+@pytest.mark.parametrize("mode", ["full", "compact", "brief", "thin", "envelope"])
+@pytest.mark.parametrize("profile", ["generic_cli", "codex_app_heartbeat"])
+def test_timed_wait_guidance_survives_prompt_compaction(mode: str, profile: str, tmp_path: Path) -> None:
+    options = ({"thin": True, "turn_envelope": True, "decision_output_root": tmp_path,
+                "turn_instance_id": "timed-wait-turn"} if mode == "envelope" else {mode: True})
+    payload = build_heartbeat_prompt(goal_id="timed-wait-fixture", agent_id="agent-wait",
+                                     runtime_profile=profile, **options)
+    body = payload["task_body"]
+    # Preserve the actual lifecycle choice and execution/settlement boundaries,
+    # not an unconditional instruction to keep every kind of wait open.
+    for obligation in (
+        "No useful independent work", "save progress", "todo update --status deferred",
+        "--resume-when resume_at:", "evidenced due time", "settle original Turn",
+        "yield to host scheduler", "not model sleep/poll loops",
+        "Fresh admission on resume", "Other waits: projected monitor/continuation",
+        "No spend for waiting alone",
+    ):
+        assert obligation in body
+    assert "Wait=open" not in body and "Wait: keep open" not in body
+    assert payload["interface_budget"]["within_budget"] is True
+
+
 @pytest.mark.parametrize("mode", ["full", "compact", "brief", "thin"])
 def test_codex_app_stop_obligations_survive_prompt_compaction(mode: str) -> None:
     payload = build_heartbeat_prompt(

tests/control_plane/test_quota_settlement_cli.py

@@ -5327,7 +5327,7 @@ def test_same_turn_bound_p0_does_not_project_p1_after_p0_becomes_deferred(
         ),
         encoding="utf-8",
     )
-    _initialize_fixture_authority(project, runtime, provider=provider)
+    _initialize_fixture_authority(project, runtime, provider=provider, handoff_mode="hard_lease")
     turn_id = "turn-bound-p0-then-deferred"
     guard = (
         "quota", "should-run", "--codex-app", "--goal-id", GOAL_ID,
@@ -5336,6 +5336,11 @@ def test_same_turn_bound_p0_does_not_project_p1_after_p0_becomes_deferred(
     first_rc, first = _run_cli(registry_path, runtime, *guard, "--todo-id", TODO_ID)
     assert first_rc == 0, first
     assert first["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO_ID
+    claim_rc, claim = _run_cli(
+        registry_path, runtime, "todo", "claim", "--goal-id", GOAL_ID,
+        "--agent-id", AGENT_ID, "--claimed-by", AGENT_ID, "--todo-id", TODO_ID,
+    )
+    assert claim_rc == 0 and claim["execution_managed"] is True, json.dumps(claim)
     # This result exists before the lifecycle wait; recovery must account for
     # it without executing more work or rebinding the original Turn.
     artifact = project / "verified-before-wait.txt"
@@ -5348,8 +5353,15 @@ def test_same_turn_bound_p0_does_not_project_p1_after_p0_becomes_deferred(
         "--status", "deferred",
         "--resume-when", "resume_at:2099-01-01T00:00:00Z",
         "--reason", "Wait for the future resume condition.",
+        "--claim-operation-id", claim["claim_operation_id"],
     )
     assert update_rc == 0, update
+    # The ordinary wait transition retires execution without a manual release.
+    inspect_rc, inspected = _run_cli(
+        registry_path, runtime, "task-lease", "inspect", "--goal-id", GOAL_ID,
+        "--todo-id", TODO_ID,
+    )
+    assert inspect_rc == 0 and inspected["lease"]["status"] == "released", inspected
     replay_rc, replay = _run_cli(registry_path, runtime, *guard)
     assert replay_rc == 0, replay
     assert replay["effective_action"] == "unsettled_host_turn_recovery"