fix(scheduler): honor final attempt permission in external worker

scripts/external_scheduler_worker.py

@@ -6,7 +6,8 @@
 should run, projects a one-line public-safe status, and either:
 
   * observe-only (default): sleeps until the scheduler hint says to re-check;
-  * with --wake-cmd: runs the configured command when should_run is true.
+  * with --wake-cmd: runs when the interaction contract requires an agent attempt
+    (legacy packets without that contract use the strict should_run boolean).
 
 The generic CLI scheduler hint carries only the *initial* interval plus a
 progression ladder and an unchanged-poll limit. Unlike codex-app's stateful
@@ -112,7 +113,6 @@ def parse_tick(payload: dict[str, Any]) -> TickDecision:
     action = str(hint.get("action") or "").strip()
     cadence_class = str(hint.get("cadence_class") or "").strip()
     reason = str(hint.get("reason") or payload.get("state") or "").strip()
-    should_run = payload.get("should_run")
     if _local_scheduler_directive(hint) == LOCAL_SCHEDULER_STOP_DIRECTIVE:
         # Terminal packets intentionally omit cold-path cadence detail: no
         # further wake is legal, so an interval cannot affect the decision.
@@ -132,12 +132,22 @@ def parse_tick(payload: dict[str, Any]) -> TickDecision:
             reset_token=_extract_reset_token(payload),
             terminal=True,
         )
-    # Consume the Core boolean, never Python truthiness (e.g. "false" or 1).
-    # The terminal directive above wins even over a malformed admission value.
+    # The final interaction contract owns wake permission. Raw should_run is
+    # advisory: it can remain true during quiet waits or false for required
+    # repair. Consume the existing projection without reclassifying its mode.
+    # A present but malformed contract must never fall back to legacy admission.
+    # The terminal directive above still wins over malformed admission values.
     if payload.get("ok") is False:
         raise ValueError("quota decision failed; refusing model wake")
+    if "interaction_contract" in payload:
+        contract = _mapping(payload["interaction_contract"])
+        should_run = _mapping(contract.get("agent_channel")).get("must_attempt")
+        admission_field = "interaction_contract.agent_channel.must_attempt"
+    else:
+        should_run = payload.get("should_run")
+        admission_field = "quota should_run"
     if not isinstance(should_run, bool):
-        raise ValueError("quota should_run must be a boolean; refusing model wake")
+        raise ValueError(f"{admission_field} must be a boolean; refusing model wake")
     local = _extract_local_scheduler(payload)
 
     progression_values = local.get("example_progression_minutes")

tests/test_external_scheduler_worker.py

@@ -185,6 +185,85 @@ def test_failed_decision_cannot_authorize_wake(tmp_path, monkeypatch):
     assert worker.run_worker(args) == 2
 
 
+@pytest.mark.parametrize("contract", [
+    None, [], {}, {"agent_channel": None}, {"agent_channel": {}},
+    *({"agent_channel": {"must_attempt": value}} for value in ("true", "false", 1, 0)),
+])
+def test_malformed_contract_cannot_fall_back_to_legacy_wake(tmp_path, contract):
+    payload = _hint_payload(should_run=True) | {"interaction_contract": contract}
+    cli = tmp_path / "quota"
+    _write_executable(cli, f"#!{sys.executable}\nprint({json.dumps(payload)!r})\n")
+    marker = tmp_path / "model-called"
+    args = _args(fake_cli=cli, state_file=tmp_path / "state.json",
+        wake_cmd=shlex.join([sys.executable, "-c",
+            f"from pathlib import Path; Path({str(marker)!r}).touch()"]),
+        wake_timeout_seconds=5)
+    assert worker.run_worker(args) == 2
+    assert not marker.exists()
+
+
+@pytest.mark.parametrize("repair", [False, True])
+def test_typed_wait_backs_off_then_fresh_attempt_wakes_once(tmp_path, repair):
+    from loopx.control_plane.scheduler.execution_context import (
+        scheduler_execution_context_for_runtime_profile,
+    )
+    from tests.control_plane.scheduler_arbitration_fixture import build_native_scheduler_hint
+
+    def packet(mode, raw, must_attempt, delivery, quiet):
+        payload = {
+            "ok": True, "should_run": raw, "effective_action": mode,
+            "interaction_contract": {
+                "schema_version": "loopx_interaction_contract_v0", "mode": mode,
+                # A notice cannot itself authorize a wake, and DONT_NOTIFY must
+                # not suppress a required attempt (including repair-only work).
+                "user_channel": {"action_required": False,
+                                 "notify": "DONT_NOTIFY" if must_attempt else "NOTIFY"},
+                "agent_channel": {"must_attempt": must_attempt,
+                                  "delivery_allowed": delivery, "quiet_noop_allowed": quiet},
+                "cli_channel": {"next_cli_actions": [], "spend_allowed_now": False},
+            },
+        }
+        payload["scheduler_hint"] = build_native_scheduler_hint(payload,
+            scheduler_execution_context=scheduler_execution_context_for_runtime_profile("generic_cli"),
+            include_detail=True)
+        return payload
+
+    waiting = packet("mapped_noop_if_unchanged", True, False, False, True)
+    active = packet("self_repair" if repair else "bounded_delivery",
+                    not repair, True, not repair, False)
+    terminal = _hint_payload(should_run=True, local_scheduler_directive="stop")
+    # Producer stop is authoritative even when the execution field is malformed.
+    terminal["interaction_contract"] = {"agent_channel": {"must_attempt": "true"}}
+    packets = [waiting, waiting, waiting, active, terminal]
+    cli, counter, marker = (tmp_path / name for name in ("quota", "quota-count", "wake-count"))
+    _write_executable(cli, f"#!{sys.executable}\n"
+        "import json\nfrom pathlib import Path\n"
+        f"counter = Path({str(counter)!r})\n"
+        "n = int(counter.read_text()) if counter.exists() else 0\n"
+        "counter.write_text(str(n + 1))\n"
+        f"print(json.dumps({packets!r}[n]))\n")
+    wake_code = ("from pathlib import Path; "
+        f"p=Path({str(marker)!r}); "
+        "p.write_text(str(int(p.read_text()) + 1) if p.exists() else '1')")
+    args = _args(fake_cli=cli, state_file=tmp_path / "state.json",
+        wake_cmd=shlex.join([sys.executable, "-c", wake_code]),
+        quota_timeout_seconds=5, wake_timeout_seconds=5)
+    args.once = False
+    sleeps = []
+
+    def wait(seconds):
+        assert not marker.exists(), "a quiet tick launched a child before backing off"
+        sleeps.append(seconds)
+
+    assert worker.run_worker(args, sleep=wait) == 0
+    # Native generic-CLI unchanged policy: two waits, then a final fresh probe.
+    # Required work found by that probe wakes immediately; successful completion
+    # rechecks the stop packet without inserting another active-work delay.
+    assert sleeps == [3600, 7200]
+    assert counter.read_text() == "5"
+    assert marker.read_text() == "1"
+
+
 def test_explicit_stop_precedes_even_malformed_admission(tmp_path, monkeypatch):
     payload = _hint_payload(should_run=True, local_scheduler_directive="stop")
     payload["should_run"] = "false"