Compact quota scheduler hints

docs/dashboard-budget-governance-contract.md

@@ -12,8 +12,8 @@ creating a browser-side source of truth.
 | Operator concept | Source fields | Meaning in the dashboard |
 | --- | --- | --- |
 | Budget | `quota.compute`, `quota.allowed_slots`, `quota.spent_slots`, `quota.state` | How much automatic agent time this goal may consume in the current quota window, and whether it can run now. |
-| Cadence | `scheduler_hint.codex_app`, `scheduler_hint.local_scheduler`, `scheduler_hint.reset_policy` | How often the host should wake the agent, when backoff applies, and when user feedback or new work resets the interval. |
-| Spend rule | `interaction_contract.cli_channel.spend_policy`, `scheduler_hint.*.no_spend_*`, `work_lane_contract` | Which transitions spend quota and which lifecycle checks are no-spend. |
+| Cadence | `scheduler_hint.codex_app`, `scheduler_hint.unchanged_poll`, `scheduler_hint.reset_policy`; opt-in cold detail from `scheduler_hint.cold_path_detail.local_scheduler` | How often the host should wake the agent, when backoff applies, and when user feedback or new work resets the interval. |
+| Spend rule | `interaction_contract.cli_channel.spend_policy`, `scheduler_hint.unchanged_poll.spend_policy`, `work_lane_contract` | Which transitions spend quota and which lifecycle checks are no-spend. |
 | Human controls | user todos, operator gates, `local_dashboard_api`, future control-plane dry-run/apply paths | What a human can approve, pause, override, or resume, and whether the browser is allowed to preview or apply a change. |
 | Evidence | todo ids, run ids, quota spend events, compact artifacts, source warnings | Why the dashboard believes the current budget/governance state and where to audit it. |
 

examples/codex-cli-local-scheduler-tick-smoke.py

@@ -115,23 +115,25 @@
         "schema_version": "scheduler_hint_v0",
         "action": "backoff_until_reassigned",
         "cadence_class": "agent_scope_wait",
-        "local_scheduler": {
+        "codex_app": {
             "recommended_interval_minutes": 10,
             "max_interval_minutes": 60,
+            "unchanged_poll_backoff_multiplier": 2,
             "example_progression_minutes": [10, 20, 30, 60],
-            "unchanged_poll_limit": 3,
-            "after_limit": "stop_tick_loop",
-            "final_quota_replan_check": {"enabled": True},
         },
-        "codex_cli_tui": {
-            "unchanged_poll_limit": 3,
-            "after_limit": "exit_goal_loop",
-            "final_quota_replan_check": {"enabled": True},
-        },
-        "claude_code_loop": {
-            "unchanged_poll_limit": 3,
-            "after_limit": "stop_loop",
-            "final_quota_replan_check": {"enabled": True},
+        "unchanged_poll": {
+            "limits": {
+                "local_scheduler": 3,
+                "codex_cli_tui": 3,
+                "claude_code_loop": 3,
+            },
+            "after_limits": {
+                "local_scheduler": "stop_tick_loop",
+                "codex_cli_tui": "exit_goal_loop",
+                "claude_code_loop": "stop_loop",
+            },
+            "final_quota_replan_check_enabled": True,
+            "final_quota_replan_check_action": "rerun_quota_should_run_once",
         },
         "reset_policy": {
             "schema_version": "scheduler_reset_policy_v0",
@@ -237,8 +239,8 @@ def main() -> int:
     assert hinted_tick["launchd"]["reset_token"] == "fixture-reset-001", hinted_tick
     assert hinted_tick["launchd"]["reset_interval_seconds"] == 600, hinted_tick
     assert hinted_tick["launchd"]["reset_policy"]["codex_app_initial_rrule"] == "FREQ=MINUTELY;INTERVAL=10", hinted_tick
-    assert hinted_tick["scheduler_hint"]["local_scheduler"]["example_progression_minutes"] == [10, 20, 30, 60], hinted_tick
-    assert hinted_tick["scheduler_hint"]["codex_cli_tui"]["final_quota_replan_check"]["enabled"] is True, hinted_tick
+    assert hinted_tick["scheduler_hint"]["codex_app"]["example_progression_minutes"] == [10, 20, 30, 60], hinted_tick
+    assert hinted_tick["scheduler_hint"]["unchanged_poll"]["final_quota_replan_check_enabled"] is True, hinted_tick
 
     with tempfile.TemporaryDirectory(prefix="loopx-codex-cli-scheduler-tick-") as tmp:
         tmp_path = Path(tmp)

examples/heartbeat-quota-flow-smoke.py

@@ -664,9 +664,13 @@ def main() -> int:
         assert first_guard["scheduler_hint"]["codex_app"]["recommended_interval_minutes"] == 15, first_guard
         assert first_guard["scheduler_hint"]["codex_app"]["recommended_rrule"] == "FREQ=MINUTELY;INTERVAL=15", first_guard
         assert first_guard["scheduler_hint"]["codex_app"]["example_progression_minutes"] == [15, 30, 60], first_guard
-        assert first_guard["scheduler_hint"]["local_scheduler"]["max_interval_minutes"] == 60, first_guard
-        assert first_guard["scheduler_hint"]["codex_cli_tui"]["unchanged_poll_limit"] == 3, first_guard
-        assert first_guard["scheduler_hint"]["codex_cli_tui"]["final_quota_replan_check"]["enabled"] is True, first_guard
+        assert first_guard["scheduler_hint"]["codex_app"]["max_interval_minutes"] == 60, first_guard
+        assert first_guard["scheduler_hint"]["unchanged_poll"]["limits"]["codex_cli_tui"] == 3, first_guard
+        assert first_guard["scheduler_hint"]["unchanged_poll"]["final_quota_replan_check_enabled"] is True, first_guard
+        assert "local_scheduler" not in first_guard["scheduler_hint"], first_guard
+        assert "codex_cli_tui" not in first_guard["scheduler_hint"], first_guard
+        assert "claude_code_loop" not in first_guard["scheduler_hint"], first_guard
+        assert "cold_path_detail" not in first_guard["scheduler_hint"], first_guard
         reset = first_guard["scheduler_hint"]["reset_policy"]
         assert reset["schema_version"] == "scheduler_reset_policy_v0", reset
         assert reset["profile_action"] == "backoff_until_material_transition", reset

examples/quota-agent-scoped-user-gate-smoke.py

@@ -505,10 +505,14 @@ def assert_agent_without_advancement_candidate_enters_scope_wait() -> None:
     assert scheduler["codex_app"]["recommended_interval_minutes"] == 10, scheduler
     assert scheduler["codex_app"]["recommended_rrule"] == "FREQ=MINUTELY;INTERVAL=10", scheduler
     assert scheduler["codex_app"]["example_progression_minutes"] == [10, 20, 30, 60], scheduler
-    assert scheduler["codex_cli_tui"]["unchanged_poll_limit"] == 3, scheduler
-    assert scheduler["codex_cli_tui"]["final_quota_replan_check"]["enabled"] is True, scheduler
-    assert scheduler["claude_code_loop"]["after_limit"] == "stop_loop", scheduler
-    assert scheduler["claude_code_loop"]["unchanged_poll_limit"] == 3, scheduler
+    assert scheduler["unchanged_poll"]["limits"]["codex_cli_tui"] == 3, scheduler
+    assert scheduler["unchanged_poll"]["final_quota_replan_check_enabled"] is True, scheduler
+    assert scheduler["unchanged_poll"]["after_limits"]["claude_code_loop"] == "stop_loop", scheduler
+    assert scheduler["unchanged_poll"]["limits"]["claude_code_loop"] == 3, scheduler
+    assert "local_scheduler" not in scheduler, scheduler
+    assert "codex_cli_tui" not in scheduler, scheduler
+    assert "claude_code_loop" not in scheduler, scheduler
+    assert "cold_path_detail" not in scheduler, scheduler
     reset = scheduler["reset_policy"]
     assert reset["schema_version"] == "scheduler_reset_policy_v0", reset
     assert reset["profile_action"] == "backoff_until_reassigned", reset

examples/quota-plan-smoke.py

@@ -49,16 +49,16 @@ def _nested_value(payload: dict, path: str):
 
 def scheduler_reset_profile_snapshot(scheduler: dict) -> dict:
     codex_app = scheduler["codex_app"]
-    local_scheduler = scheduler["local_scheduler"]
-    claude_code_loop = scheduler["claude_code_loop"]
+    unchanged_poll = scheduler["unchanged_poll"]
+    limits = unchanged_poll["limits"]
     return {
         "cadence_class": scheduler["cadence_class"],
         "codex_app_initial_interval_minutes": codex_app["recommended_interval_minutes"],
         "codex_app_initial_rrule": codex_app["recommended_rrule"],
         "codex_app_max_interval_minutes": codex_app["max_interval_minutes"],
         "unchanged_poll_backoff_multiplier": codex_app["unchanged_poll_backoff_multiplier"],
-        "local_scheduler_unchanged_poll_limit": local_scheduler["unchanged_poll_limit"],
-        "claude_code_loop_unchanged_poll_limit": claude_code_loop["unchanged_poll_limit"],
+        "local_scheduler_unchanged_poll_limit": limits["local_scheduler"],
+        "claude_code_loop_unchanged_poll_limit": limits["claude_code_loop"],
     }
 
 
@@ -1604,8 +1604,12 @@ def assert_heartbeat_recommendation_lifecycle() -> None:
     assert scheduler["codex_app"]["recommended_interval_minutes"] == 60, mapped_decision
     assert scheduler["codex_app"]["recommended_rrule"] == "FREQ=MINUTELY;INTERVAL=60", mapped_decision
     assert scheduler["codex_app"]["example_progression_minutes"] == [60, 120, 240], mapped_decision
-    assert scheduler["codex_cli_tui"]["unchanged_poll_limit"] == 3, mapped_decision
-    assert scheduler["codex_cli_tui"]["final_quota_replan_check"]["enabled"] is True, mapped_decision
+    assert scheduler["unchanged_poll"]["limits"]["codex_cli_tui"] == 3, mapped_decision
+    assert scheduler["unchanged_poll"]["final_quota_replan_check_enabled"] is True, mapped_decision
+    assert "local_scheduler" not in scheduler, scheduler
+    assert "codex_cli_tui" not in scheduler, scheduler
+    assert "claude_code_loop" not in scheduler, scheduler
+    assert "cold_path_detail" not in scheduler, scheduler
     reset = scheduler["reset_policy"]
     assert reset["schema_version"] == "scheduler_reset_policy_v0", reset
     assert reset["reset_to"] == "profile_initial_interval", reset

examples/quota-scheduler-hint-compaction-smoke.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python3
+"""Smoke-test the compact quota scheduler_hint hot-path contract."""
+
+from __future__ import annotations
+
+from copy import deepcopy
+import json
+from pathlib import Path
+import sys
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from loopx.policies.scheduler_hint import build_scheduler_hint  # noqa: E402
+from loopx.quota import _scheduler_hint  # noqa: E402
+
+
+def payload(*, should_run: bool, recommended_mode: str = "", user_required: bool = False) -> dict:
+    return {
+        "goal_id": "quota-scheduler-compaction",
+        "should_run": should_run,
+        "effective_action": "operator_gate_notify" if user_required else "normal_run",
+        "recommended_action": "Keep scheduler hints compact on the hot path.",
+        "heartbeat_recommendation": {
+            "recommended_mode": recommended_mode,
+            "notify": "NOTIFY" if user_required else "DONT_NOTIFY",
+            "spend_policy": "spend only after validated writeback",
+        },
+        "execution_obligation": {
+            "must_attempt_work": should_run,
+            "spend_policy": "execution obligation spend policy",
+        },
+        "automation_liveness": {
+            "automation_action": "",
+            "spend_policy": "automation liveness spend policy",
+        },
+        "interaction_contract": {
+            "mode": recommended_mode or "normal_run",
+            "user_channel": {
+                "action_required": user_required,
+            },
+        },
+    }
+
+
+def json_size(value: dict) -> int:
+    return len(json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")))
+
+
+def assert_compact_scheduler(name: str, source_payload: dict) -> None:
+    compact = build_scheduler_hint(deepcopy(source_payload), user_action_required=False)
+    wrapper = _scheduler_hint(deepcopy(source_payload))
+    detailed = build_scheduler_hint(
+        deepcopy(source_payload),
+        user_action_required=False,
+        include_detail=True,
+    )
+
+    assert compact == wrapper, (name, compact, wrapper)
+    assert compact["schema_version"] == "scheduler_hint_v0", (name, compact)
+    assert "local_scheduler" not in compact, (name, compact)
+    assert "codex_cli_tui" not in compact, (name, compact)
+    assert "claude_code_loop" not in compact, (name, compact)
+    assert "cold_path_detail" not in compact, (name, compact)
+    assert compact["detail_ref"]["omitted_by_default"] is True, (name, compact)
+    assert compact["detail_ref"]["request"] == "loopx quota should-run --include-scheduler-detail", (name, compact)
+    assert compact["reset_policy"]["reset_token"], (name, compact)
+    assert compact["reset_policy"]["codex_app_initial_rrule"] == compact["codex_app"]["recommended_rrule"], (
+        name,
+        compact,
+    )
+    assert "identity_snapshot" not in compact["reset_policy"], (name, compact)
+    assert "profile_snapshot" not in compact["reset_policy"], (name, compact)
+
+    unchanged_poll = compact["unchanged_poll"]
+    assert isinstance(unchanged_poll["limits"], dict), (name, compact)
+    assert isinstance(unchanged_poll["after_limits"], dict), (name, compact)
+    assert "final_quota_replan_check" not in unchanged_poll, (name, compact)
+
+    cold_path = detailed["cold_path_detail"]
+    assert cold_path["schema_version"] == "scheduler_hint_detail_v0", (name, detailed)
+    assert cold_path["local_scheduler"]["recommended_interval_minutes"], (name, detailed)
+    assert cold_path["codex_cli_tui"]["final_quota_replan_check"], (name, detailed)
+    assert cold_path["claude_code_loop"]["after_limit"], (name, detailed)
+    assert json_size(compact) < json_size(detailed), (name, json_size(compact), json_size(detailed))
+    assert json_size(compact) <= 2_800, (name, json_size(compact))
+
+
+def main() -> int:
+    assert_compact_scheduler("active-work", payload(should_run=True))
+    assert_compact_scheduler(
+        "human-gate",
+        payload(should_run=False, recommended_mode="ask_operator_gate", user_required=True),
+    )
+    print("quota-scheduler-hint-compaction-smoke ok")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

examples/quota-scheduler-policy-smoke.py

@@ -67,11 +67,29 @@ def assert_policy_case(name: str, base_payload: dict, *, expected_action: str, e
         ),
         agent_scope_frontier_actions=AGENT_SCOPE_ACTIONS,
     )
+    detailed = build_scheduler_hint(
+        deepcopy(base_payload),
+        user_action_required=bool(
+            base_payload.get("interaction_contract", {})
+            .get("user_channel", {})
+            .get("action_required")
+        ),
+        agent_scope_frontier_actions=AGENT_SCOPE_ACTIONS,
+        include_detail=True,
+    )
     assert extracted == quota_wrapper, (name, extracted, quota_wrapper)
     assert extracted["schema_version"] == "scheduler_hint_v0", (name, extracted)
     assert extracted["source"] == "quota.should-run", (name, extracted)
     assert extracted["action"] == expected_action, (name, extracted)
     assert extracted["codex_app"]["recommended_rrule"] == expected_rrule, (name, extracted)
+    assert "local_scheduler" not in extracted, (name, extracted)
+    assert "codex_cli_tui" not in extracted, (name, extracted)
+    assert "claude_code_loop" not in extracted, (name, extracted)
+    assert "cold_path_detail" not in extracted, (name, extracted)
+    assert extracted["detail_ref"]["request"] == "loopx quota should-run --include-scheduler-detail", (name, extracted)
+    assert detailed["cold_path_detail"]["local_scheduler"]["recommended_interval_minutes"], (name, detailed)
+    assert detailed["cold_path_detail"]["codex_cli_tui"]["final_quota_replan_check"], (name, detailed)
+    assert detailed["cold_path_detail"]["claude_code_loop"]["after_limit"], (name, detailed)
     reset = extracted["reset_policy"]
     assert reset["schema_version"] == "scheduler_reset_policy_v0", (name, reset)
     assert isinstance(reset["reset_token"], str) and len(reset["reset_token"]) == 16, (name, reset)

loopx/cli_commands/quota.py

@@ -67,6 +67,14 @@ def register_quota_command(subparsers: argparse._SubParsersAction) -> None:
             "capabilities; basic local shell/filesystem capabilities are assumed."
         ),
     )
+    quota_parser.add_argument(
+        "--include-scheduler-detail",
+        action="store_true",
+        help=(
+            "Include cold-path scheduler detail for local scheduler, Codex CLI, "
+            "and Claude loop runtimes in `quota should-run` JSON."
+        ),
+    )
     quota_parser.add_argument("--slots", type=int, default=1, help="Slots to account for `quota spend-slot`.")
     quota_parser.add_argument("--source", choices=["heartbeat", "controller", "adapter"], default="heartbeat", help="Source label for `quota spend-slot`.")
     quota_parser.add_argument("--void-generated-at", help="generated_at timestamp of the quota_slot_spent run to void.")
@@ -125,6 +133,7 @@ def handle_quota_command(
                 goal_id=args.goal_id,
                 agent_id=args.agent_id,
                 available_capabilities=args.available_capabilities,
+                include_scheduler_detail=bool(args.include_scheduler_detail),
             )
         elif args.quota_command == "monitor-poll":
             if not args.goal_id:

loopx/codex_cli_probe.py

@@ -1388,13 +1388,19 @@ def build_codex_cli_local_scheduler_tick(
         if isinstance(scheduler_hint.get("local_scheduler"), dict)
         else {}
     )
+    codex_app_hint = (
+        scheduler_hint.get("codex_app")
+        if isinstance(scheduler_hint.get("codex_app"), dict)
+        else {}
+    )
     reset_policy = (
         scheduler_hint.get("reset_policy")
         if isinstance(scheduler_hint.get("reset_policy"), dict)
         else {}
     )
     recommended_interval_minutes = _positive_int(
-        local_scheduler_hint.get("recommended_interval_minutes"),
+        local_scheduler_hint.get("recommended_interval_minutes")
+        or codex_app_hint.get("recommended_interval_minutes"),
         10,
     )
     reset_interval_minutes = _positive_int(
@@ -2849,6 +2855,46 @@ def render_codex_cli_local_scheduler_tick_markdown(payload: dict[str, Any]) -> s
         if isinstance(scheduler_hint.get("local_scheduler"), dict)
         else {}
     )
+    codex_app = (
+        scheduler_hint.get("codex_app")
+        if isinstance(scheduler_hint.get("codex_app"), dict)
+        else {}
+    )
+    unchanged_poll = (
+        scheduler_hint.get("unchanged_poll")
+        if isinstance(scheduler_hint.get("unchanged_poll"), dict)
+        else {}
+    )
+    limits = unchanged_poll.get("limits") if isinstance(unchanged_poll.get("limits"), dict) else {}
+    after_limits = (
+        unchanged_poll.get("after_limits")
+        if isinstance(unchanged_poll.get("after_limits"), dict)
+        else {}
+    )
+    local_interval = (
+        local_scheduler.get("recommended_interval_minutes")
+        or codex_app.get("recommended_interval_minutes")
+    )
+    local_progression = (
+        local_scheduler.get("example_progression_minutes")
+        or codex_app.get("example_progression_minutes")
+    )
+    local_unchanged_limit = (
+        local_scheduler.get("unchanged_poll_limit")
+        if "unchanged_poll_limit" in local_scheduler
+        else limits.get("local_scheduler")
+    )
+    local_after_limit = (
+        local_scheduler.get("after_limit")
+        or after_limits.get("local_scheduler")
+    )
+    final_replan_check = (
+        local_scheduler.get("final_quota_replan_check")
+        or {
+            "enabled": unchanged_poll.get("final_quota_replan_check_enabled"),
+            "action": unchanged_poll.get("final_quota_replan_check_action"),
+        }
+    )
     blocker = payload.get("precise_blocker") if isinstance(payload.get("precise_blocker"), dict) else None
     runtime_idle = (
         payload.get("runtime_idle_detector")
@@ -2900,11 +2946,11 @@ def render_codex_cli_local_scheduler_tick_markdown(payload: dict[str, Any]) -> s
 
 - action: `{scheduler_hint.get("action")}`
 - cadence_class: `{scheduler_hint.get("cadence_class")}`
-- local_interval_minutes: `{local_scheduler.get("recommended_interval_minutes")}`
-- local_progression_minutes: `{local_scheduler.get("example_progression_minutes")}`
-- local_unchanged_poll_limit: `{local_scheduler.get("unchanged_poll_limit")}`
-- local_after_limit: `{local_scheduler.get("after_limit")}`
-- final_quota_replan_check: `{local_scheduler.get("final_quota_replan_check")}`
+- local_interval_minutes: `{local_interval}`
+- local_progression_minutes: `{local_progression}`
+- local_unchanged_poll_limit: `{local_unchanged_limit}`
+- local_after_limit: `{local_after_limit}`
+- final_quota_replan_check: `{final_replan_check}`
 
 ## Boundary