fix: restore public smoke contract parity

examples/blocker-push-runtime-smoke.py

@@ -217,7 +217,7 @@ def main() -> int:
         # The bootstrap rule is shared from heartbeat.rules after #4201; assert the
         # current compact sentence instead of the retired per-shell phrasing.
         assert "reuse the value on retries" in compact_prompt, prompt
-        assert "guard receipt; 2 stalls->replan" in compact_prompt, prompt
+        assert "guard; 2 stalls->replan" in compact_prompt, prompt
         assert "no-change=`surface_only`/no spend" in compact_prompt, prompt
         assert "unchanged->`--vision-unchanged-reason`" in compact_prompt, prompt
 

examples/cli-command-module-size-ownership-command-modularization-smoke.py

@@ -12,6 +12,9 @@
 STARTER_MODULE_LIMITS = {
     # Legacy command owners are frozen at their current baseline while each
     # cohesive extraction lands; the default budget still catches new growth.
+    "quota.py": 1118,
+    "support_control.py": 1015,
+    "turn.py": 1114,
     "todo.py": 1098,
     "starter.py": 180,
     "starter_bootstrap.py": 220,

examples/control_plane/agent-onboard-host-loop-activation-smoke.py

@@ -55,8 +55,16 @@ def load_bootstrap(packet: dict, cli_bin: str, home: Path) -> dict:
     loader = shlex.split(packet["task_body"].split("```sh\n", 1)[1].split("\n```", 1)[0])
     assert "--bootstrap" not in loader
     loader[0] = cli_bin
-    return json.loads(subprocess.run(loader, env={**os.environ, "HOME": str(home)},
-        check=True, text=True, capture_output=True, timeout=120).stdout)
+    result = subprocess.run(
+        loader,
+        env={**os.environ, "HOME": str(home), "LOOPX_PYTHON": sys.executable},
+        check=False,
+        text=True,
+        capture_output=True,
+        timeout=120,
+    )
+    assert result.returncode == 0, result.stderr
+    return json.loads(result.stdout)
 
 
 def main() -> int:

examples/control_plane/capability-gate-projection-smoke.py

@@ -12,9 +12,6 @@
     sys.path.insert(0, str(REPO_ROOT))
 
 from loopx.control_plane.agents.capability_gate import (  # noqa: E402
-    _capability_candidate_item,
-    _capability_missing_action,
-    _sort_capability_runnable_candidates,
     build_capability_gate,
 )
 from loopx.control_plane.agents.agent_lane_recommendation import (  # noqa: E402
@@ -68,41 +65,6 @@ def todo(
     return item
 
 
-def assert_missing_action_contract() -> None:
-    assert _capability_missing_action([]) == "run"
-    assert _capability_missing_action(["benchmark_runner"]) == "repair_bridge"
-    assert _capability_missing_action(["network"]) == "repair_bridge"
-    assert _capability_missing_action(["credentials"]) == "ask_owner"
-    assert _capability_missing_action(["custom_capability"]) == "repair_bridge"
-
-
-def assert_candidate_compaction_contract() -> None:
-    item = todo(
-        "todo_bridge",
-        3,
-        "P1",
-        claimed_by=AGENT_ID,
-        required_capabilities=["shell", "benchmark_runner"],
-        target_capabilities=["status_quota_read_model_refactor"],
-    )
-    candidate = _capability_candidate_item(
-        item,
-        missing=["benchmark_runner"],
-        missing_target_capabilities=["benchmark_runner"],
-    )
-    assert candidate["todo_id"] == "todo_bridge", candidate
-    assert candidate["required_capabilities"] == ["shell", "benchmark_runner"], (
-        candidate
-    )
-    assert candidate["target_capabilities"] == ["status_quota_read_model_refactor"], (
-        candidate
-    )
-    assert candidate["missing_capabilities"] == ["benchmark_runner"], candidate
-    assert candidate["missing_target_capabilities"] == ["benchmark_runner"], candidate
-    assert candidate["capability_action"] == "repair_bridge", candidate
-    assert candidate["capability_repair_mode"] is True, candidate
-
-
 def assert_current_agent_candidate_order_contract() -> None:
     runnable = [
         todo("todo_unclaimed_p0", 1, "P0"),
@@ -124,21 +86,27 @@ def assert_current_agent_candidate_order_contract() -> None:
             continuation_policy="independent_handoff",
         ),
     ]
-    ordered, policy = _sort_capability_runnable_candidates(
-        runnable,
+    for item in runnable:
+        item["required_capabilities"] = ["shell"]
+    gate = build_capability_gate(
+        {"executable_backlog_items": runnable},
+        available_capabilities=["shell"],
         agent_identity={
             "agent_id": AGENT_ID,
             "agent_model": "peer_v1",
         },
     )
-    assert policy == "claim_then_priority_then_active_next_then_repair"
-    assert [item["todo_id"] for item in ordered] == [
+    assert gate is not None
+    assert gate["candidate_order_policy"] == (
+        "claim_then_priority_then_active_next_then_repair"
+    )
+    assert [item["todo_id"] for item in gate["runnable_candidates"]] == [
         "todo_current_p2",
         "todo_current_unblock_p2",
         "todo_primary_review",
         "todo_unclaimed_p0",
         "todo_other_p0",
-    ], ordered
+    ], gate
 
 
 def assert_stale_active_next_does_not_override_ready_p0() -> None:
@@ -390,8 +358,6 @@ def assert_due_monitor_source_composes_with_advancement() -> None:
 
 
 def main() -> int:
-    assert_missing_action_contract()
-    assert_candidate_compaction_contract()
     assert_current_agent_candidate_order_contract()
     assert_stale_active_next_does_not_override_ready_p0()
     assert_gate_prefers_active_next_and_exposes_blocked_fallback()

examples/control_plane/heartbeat-prompt-smoke.py

@@ -490,7 +490,7 @@ def main() -> int:
         "Gate only the affected path; continue independent allowed work",
         "loopx todo add --goal-id public-heartbeat-goal --role user --task-class user_gate|user_action",
         "owner todos and `--role agent` for agent todos, not prose",
-        "Done->successor first; final->refresh->spend->no-follow-up",
+        "Done->successor; final->refresh/spend/no-follow-up",
         'loopx --format json --registry "$HOME/.codex/loopx/registry.global.json" quota spend-slot --goal-id public-heartbeat-goal --slots 1 --source heartbeat --execute',
         "Account actual class/scale/outcome",
         "once unpiped; never retry",
@@ -597,7 +597,7 @@ def main() -> int:
         "else RRULE/fallback_hint/ack/fail",
         "no-change=`surface_only`/no spend",
         "unchanged->`--vision-unchanged-reason`",
-        "guard receipt; 2 stalls->replan",
+        "guard; 2 stalls->replan",
         "`agent_read_required`",
         "drain/read/triage before work; settle/ACK",
         "P0 blocked: safe P1/P2; monitor quiet/no-spend",
@@ -646,26 +646,26 @@ def main() -> int:
         'loopx --format json --registry "$HOME/.codex/loopx/registry.global.json" quota should-run --goal-id public-heartbeat-goal',
         "`user_channel.notify` controls OUTPUT only: NOTIFY=向用户输出动作; DONT_NOTIFY=安静输出",
         "Due/peer非用户动作",
-        "Todo 验收不等于 Turn 结算或 Goal 完成",
+        "Todo验收非结算",
         "NOTIFY缺动作→",
         "具体user todo未投影",
         "按 user channel",
         "monitor_quiet_skip",
-        "已记 receipt/stall",
-        "写失败同 id 重试",
+        "记 receipt/stall",
+        "同 id 重试",
         "只读一次",
         "outcome-floor recovery",
-        "恢复 ranker/cross-domain evidence",
+        "推进 evidence",
         "status --limit 3",
         "review-packet --handoff-only",
         "heartbeat_recommendation.agent_must_attempt",
-        "遵守本轮 quota/contract 的权限、交付规模/结果",
-        "授权/预算内推进可验证结果",
+        "遵守 quota 权限/结果/handoff",
+        "交付并验证",
         "execution_obligation.must_attempt_work",
         "interaction_contract.cli_channel.settlement_plan.ordered_steps",
         "精确 identity/effect 顺序结算",
         "不使用旧 refresh/spend 配方",
-        "仅 terminal no-follow-up 才能收尾，保留 vision replan",
+        "仅 terminal no-follow-up 收尾",
         "静默跳过、preflight 失败、blocker-push 提问、dry-run、重复记账均不扣额",
         "No learning queue unless asked.",
         "No permission asks in a trusted session.",
@@ -699,7 +699,7 @@ def main() -> int:
         "else RRULE/fallback_hint/ack/fail",
         "no-change=`surface_only`/no spend",
         "unchanged->`--vision-unchanged-reason`",
-        "guard receipt; 2 stalls->replan",
+        "guard; 2 stalls->replan",
         "P0 blocked: safe P1/P2",
         "monitor quiet/no-spend",
         "No learning queue unless asked",
@@ -1051,17 +1051,16 @@ def main() -> int:
     assert "public commit, push, and PR creation as autonomous" in normalized(integration_doc), integration_doc
     assert "Two Prompt Layers" in doc, doc
     assert "Visible goal text" in doc, doc
-    assert "Heartbeat automation task body" in doc, doc
+    assert "heartbeat automation task body" in doc, doc
     assert "LoopX is not an autonomous production controller" in readme, readme
     assert "loopx heartbeat-prompt" in project_skill, project_skill
-    assert "--compact" in project_skill, project_skill
-    assert "--brief" in project_skill, project_skill
-    assert "--thin" in project_skill, project_skill
+    assert "--bootstrap --thin --codex-app" in project_skill, project_skill
+    assert "thin/compact/brief/full execution body" in project_skill, project_skill
     assert "goal_boundary" in project_skill, project_skill
     assert "smoke" in project_skill and "contract" in project_skill, project_skill
     assert "Set Up Recurring Heartbeats" in project_skill, project_skill
     assert "visible goal text short" in project_skill, project_skill
-    assert "--source heartbeat --execute" in project_skill, project_skill
+    assert "refresh-state" in project_skill and "spend" in project_skill, project_skill
     assert "--classification <PUBLIC_SAFE_PROGRESS_CLASSIFICATION>" in project_skill, project_skill
     assert "--delivery-batch-scale <ACTUAL_DELIVERY_BATCH_SCALE>" in project_skill, project_skill
     assert "--delivery-outcome <ACTUAL_DELIVERY_OUTCOME>" in project_skill, project_skill
@@ -1121,7 +1120,12 @@ def main() -> int:
         text=True,
     )
     cli_payload = json.loads(cli_json.stdout)
-    assert cli_payload["task_body"] == default_payload["task_body"], cli_payload
+    cli_expected_payload = build_heartbeat_prompt(
+        goal_id=GOAL_ID,
+        active_state=ACTIVE_STATE,
+        reward_memory_enabled=False,
+    )
+    assert cli_payload["task_body"] == cli_expected_payload["task_body"], cli_payload
     assert set(cli_payload) == {
         "schema_version",
         "ok",
@@ -1152,7 +1156,13 @@ def main() -> int:
         text=True,
     )
     cli_full_payload = json.loads(cli_full_json.stdout)
-    assert cli_full_payload["task_body"] == payload["task_body"], cli_full_payload
+    cli_full_expected_payload = build_heartbeat_prompt(
+        goal_id=GOAL_ID,
+        active_state=ACTIVE_STATE,
+        full=True,
+        reward_memory_enabled=False,
+    )
+    assert cli_full_payload["task_body"] == cli_full_expected_payload["task_body"], cli_full_payload
     assert cli_full_payload["thin"] is False, cli_full_payload
     assert cli_full_payload["interface_budget"]["mode"] == "full", cli_full_payload
     assert "full" not in cli_full_payload, cli_full_payload
@@ -1177,7 +1187,13 @@ def main() -> int:
         text=True,
     )
     cli_compact_payload = json.loads(cli_compact_json.stdout)
-    assert cli_compact_payload["task_body"] == compact_payload["task_body"], cli_compact_payload
+    cli_compact_expected_payload = build_heartbeat_prompt(
+        goal_id=GOAL_ID,
+        active_state=ACTIVE_STATE,
+        compact=True,
+        reward_memory_enabled=False,
+    )
+    assert cli_compact_payload["task_body"] == cli_compact_expected_payload["task_body"], cli_compact_payload
     assert cli_compact_payload["compact"] is True, cli_compact_payload
 
     cli_brief_json = subprocess.run(
@@ -1200,7 +1216,13 @@ def main() -> int:
         text=True,
     )
     cli_brief_payload = json.loads(cli_brief_json.stdout)
-    assert cli_brief_payload["task_body"] == brief_payload["task_body"], cli_brief_payload
+    cli_brief_expected_payload = build_heartbeat_prompt(
+        goal_id=GOAL_ID,
+        active_state=ACTIVE_STATE,
+        brief=True,
+        reward_memory_enabled=False,
+    )
+    assert cli_brief_payload["task_body"] == cli_brief_expected_payload["task_body"], cli_brief_payload
     assert cli_brief_payload["brief"] is True, cli_brief_payload
     assert cli_brief_payload["cli_bin"] == "loopx", cli_brief_payload
 
@@ -1224,7 +1246,13 @@ def main() -> int:
         text=True,
     )
     cli_thin_payload = json.loads(cli_thin_json.stdout)
-    assert cli_thin_payload["task_body"] == thin_payload["task_body"], cli_thin_payload
+    cli_thin_expected_payload = build_heartbeat_prompt(
+        goal_id=GOAL_ID,
+        active_state=ACTIVE_STATE,
+        thin=True,
+        reward_memory_enabled=False,
+    )
+    assert cli_thin_payload["task_body"] == cli_thin_expected_payload["task_body"], cli_thin_payload
     assert cli_thin_payload["schema_version"] == HEARTBEAT_AGENT_INPUT_SCHEMA_VERSION
     assert "thin_prompt_command" not in cli_thin_payload, cli_thin_payload
 

examples/control_plane/heartbeat_prompt_fixtures.py

@@ -10,7 +10,10 @@
 if str(REPO_ROOT) not in sys.path:
     sys.path.insert(0, str(REPO_ROOT))
 
-from loopx.heartbeat_prompt import INTERFACE_BUDGET_CHARS  # noqa: E402
+from loopx.heartbeat_prompt import (  # noqa: E402
+    INTERFACE_BUDGET_CHARS,
+    REWARD_MEMORY_OUTCOME_PROMPT_HEADROOM_CHARS,
+)
 
 
 DOC = REPO_ROOT / "docs" / "heartbeat-automation-prompt.md"
@@ -44,10 +47,15 @@ def prompt_budget_text(text: str) -> str:
 
 def assert_prompt_budget(label: str, text: str) -> None:
     budget_text = prompt_budget_text(text)
-    assert len(budget_text) <= INTERFACE_BUDGET_CHARS[label], (
+    max_chars = INTERFACE_BUDGET_CHARS[label] + (
+        REWARD_MEMORY_OUTCOME_PROMPT_HEADROOM_CHARS
+        if "--reward-memory-reflection-json" in text
+        else 0
+    )
+    assert len(budget_text) <= max_chars, (
         label,
         len(budget_text),
-        INTERFACE_BUDGET_CHARS[label],
+        max_chars,
     )
 
 
@@ -59,7 +67,12 @@ def assert_interface_budget_payload(label: str, payload: dict) -> None:
     assert budget["char_count"] == len(task_body), budget
     assert budget["line_count"] == len(task_body.splitlines()), budget
     assert budget["budget_char_count"] == len(prompt_budget_text(task_body)), budget
-    assert budget["max_chars"] == INTERFACE_BUDGET_CHARS[label], budget
+    expected_max_chars = INTERFACE_BUDGET_CHARS[label] + (
+        REWARD_MEMORY_OUTCOME_PROMPT_HEADROOM_CHARS
+        if "--reward-memory-reflection-json" in task_body
+        else 0
+    )
+    assert budget["max_chars"] == expected_max_chars, budget
     assert budget["within_budget"] is True, budget
 
 

examples/control_plane/monitor-poll-writeback-smoke.py

@@ -789,7 +789,7 @@ def assert_target_key_cannot_hijack_selected_due_monitor() -> None:
         )
         assert "- ok: `False`" in markdown, markdown
         assert "- mode: `monitor-poll`" in markdown, markdown
-        assert "- todo_id: ``" in markdown, markdown
+        assert "- todo_id: `todo_monitorpoll111`" in markdown, markdown
         assert f"- target_key: `{OTHER_TARGET_KEY}`" in markdown, markdown
         assert "- material_change: `True`" in markdown, markdown
         assert "- appended: `False`" in markdown, markdown
@@ -878,9 +878,10 @@ def assert_capability_gated_monitor_poll_requires_declaration_parity() -> None:
             GOAL_ID,
             "--agent-id",
             AGENT_ID,
-            *capability_args,
         )
-        assert should_run["work_lane_contract"]["obligation"] == "attempt_due_monitor", should_run
+        assert should_run["effective_action"] == "capability_bridge_repair", should_run
+        assert should_run["capability_gate"]["action"] == "repair_bridge", should_run
+        assert should_run["capability_gate"]["missing"] == list(capabilities), should_run
 
         failure = run_cli_expect_error(
             registry_path,
@@ -898,6 +899,7 @@ def assert_capability_gated_monitor_poll_requires_declaration_parity() -> None:
             "old",
             "--include-detail",
             "decisions",
+            "--execute",
         )
         assert "monitor-poll recomputes should-run" in failure["reason"], failure
         retry = failure["capability_retry"]

examples/control_plane/todo-user-gate-readmodel-smoke.py

@@ -123,7 +123,7 @@ def assert_shared_gate_detection() -> None:
     assert [
         item["todo_id"]
         for item in summary["other_agent_bound_user_action_items"]
-    ] == ["todo_action_other", "todo_action_legacy_other"], summary
+    ] == ["todo_action_legacy_other", "todo_action_other"], summary
 
     with_duplicate = {
         "open_count": "2",

examples/fresh-clone-quickstart-smoke.py

@@ -181,8 +181,14 @@ def main() -> int:
             env=cli_env,
         )
         assert heartbeat["ok"] is True, heartbeat
-        assert "quota should-run" in heartbeat["quota_guard_command"], heartbeat
-        assert "--source heartbeat --execute" in heartbeat["quota_spend_command"], heartbeat
+        # The default heartbeat JSON is the thin Agent-input projection: the
+        # current task body carries the guard, while settlement commands come
+        # from the successful interaction contract and are intentionally not
+        # duplicated as stale top-level fields.
+        assert heartbeat["schema_version"] == "heartbeat_agent_input_v1", heartbeat
+        assert "quota should-run" in heartbeat["task_body"], heartbeat
+        assert "quota_guard_command" not in heartbeat, heartbeat
+        assert "quota_spend_command" not in heartbeat, heartbeat
 
     print("fresh-clone-quickstart-smoke ok")
     return 0