diff --git a/docs/build-plans/bp-154/journal.md b/docs/build-plans/bp-154/journal.md index 8c058b94..262dfb2a 100644 --- a/docs/build-plans/bp-154/journal.md +++ b/docs/build-plans/bp-154/journal.md @@ -1,5 +1,162 @@ # bp-154 — journal +## SEAL — 2026-08-05 — all five items landed; gate recorded; PR opened for the owner's merge + +**Status.** The supervisor has a power axis. Items 1–5 are complete on +`build/bp-154-power-axis` (5 commits, `3e0713f`..`cf8d24e`); the full local gate is recorded +below; the one hard pin held (the power rule is its own predicate) and the composition falsifier +was executed rather than asserted. + +### Completed, per acceptance criterion + +- **Item 1 — the `Power` sensor** (`3e0713f`). `scheduler/power.py`, mirroring `presence.py`: an + injectable probe over `pmset -g batt`, a floor, and `assume_discharging_when_unknown = True`. + All four paths to `None` are tested (absent tool, failed/hung exec, unreadable output, a probe + that RAISES). 22 unit tests, no subprocess among them. + *Non-vacuity:* every fail-closed assertion also asserts the probe was consulted (a call log), so + none of them could pass against a sensor that never looks; and + `test_the_fail_closed_default_is_what_makes_that_true` flips the field to prove the default is + load-bearing rather than incidental. +- **Item 2 — `power_blocked_tiers()`** (`8f83561`), landed **inert**: `tick` untouched, so no + dispatch decision changed. `blocked_tiers()` is asserted byte-identical across all four + presence × power combinations, and a source-shape test pins that each predicate reads its own + sensor and no other's. + *Non-vacuity:* the source test asserts `self.power.` IS findable elsewhere in the class, so its + absence from `blocked_tiers`'s body is evidence and not a search that matches nothing. +- **Item 3 — composed at the ONE claim site** (`007324a`). One term added to the union at + `supervisor.py`; nothing else. + *Non-vacuity:* the composition test asserts, before claiming, that the foreground gate is open, + the model rule unarmed, and the floor unreached (55%) — so the power term is provably the only + rule that can refuse. **The falsifier was executed:** deleting `| self.power_blocked_tiers()` + reds exactly `test_a_heavy_job_is_NOT_claimed_while_discharging_and_IS_once_back_on_AC` + (1 failed, 52 passed) while every sensor and predicate test stays green — the finding-0187 shape + caught in the act. +- **Item 4 — the floor** (`b687031`). `tick` refuses **before** `claim` below the floor, so no + RUNNING row is minted; the test asserts the clean close as the next run experiences it + (`sweep_orphans` → nothing requeued, nothing failed). + *Non-vacuity:* the floor test enqueues a **light**-tier job, so the discharging shed provably + does not cover it — the test cannot pass on Item 3's behaviour. Deleting the floor branch reds + four tests. The "must not spin hot" falsifier is executable: a counting probe proves a ten-tick + drain request costs exactly ONE battery reading. +- **Item 5 — the carried surface** (`cf8d24e`). Six test files construct a real `Supervisor`; all + six now inject a power sensor. + *Non-vacuity:* proven necessary, not assumed — with the probe forced unreadable (the CI runner's + condition: `ubuntu-latest`, no `pmset`), `test_cron_jobs_are_gated_during_foreground_then_run_in + _a_trough` fails `assert trough.run() == 2` with 0. With the injections in place under the same + simulation, the whole model-free tier is green but for the three known-red worktree tests. + +### The gate (each leg run separately; counts exactly as observed) + +| leg | result | +|---|---| +| `uv run ruff check .` | exit 0 — All checks passed | +| `uv run mypy core agents eval ops scheduler scripts` | exit 0 — no issues in 263 source files | +| `uv run mypy` (argless) | exit 1 — **69 errors in 20 files** (the expected baseline; none in the new files) | +| `uv run python -m ops.type_gate` | exit 0 — membership OK, bare-ignore OK, one pre-existing parked `psutil` report | +| `uv run pytest -q` | exit 1 — **5 failed, 2463 passed, 15 skipped** in 278 s | + +The five reds are the three known-red classes and nothing else: the finding-0103 +core-self-containment ratchet, `tests/e2e/test_dream_v2_live.py`, and three +`tests/integration/test_worktree_enforcement.py` cases (issue #13/finding-0280 — green in CI). +⚑ `tests/e2e/test_scheduler_live.py` — in write scope, and the known flake — **passed** in the +gate run. It had failed in an earlier full run of this session, on `assert captured["text"].strip()` +with `sup.run() == 1` and `state == DONE` already passing: the job dispatched and the live model +returned an empty string. Not a power refusal (a refusal fails the `run() == 1` line one assertion +earlier), and it now carries an injected on-AC sensor. + +### Decisions taken in-build, with their warrant + +- **Unknown PERCENTAGE does not halt** (`halt_when_percent_unknown = False`, a named field, not an + omission). Fail-closed on the *discharging* question costs the heavy lanes — A1.2's rule, and the + degradation §10 already calls "safe but useless". Fail-closed on the *floor* refuses every tier, + so a host with no battery to read (a desktop, CI, any non-macOS worker) would dispatch nothing + ever: the guard against an outage would BE the outage. The knob exists, defaulted off, asserted in + both positions. +- **The hold is the absence of dispatch, not a process exit** — a deviation from A1's parked + hold-for-AC default, filed as **#40** with its evidence rather than taken silently: + `ThrottleInterval` is 10 s and every respawn pays preflight's uncosted ~120 s Ollama probe + (finding-0195), so a restart loop at the floor would spend the very reserve the floor protects. + `ops/lifecycle/launcher.py` is also outside this plan's write_scope. +- **Write-scope overrun, flagged not hidden:** `tests/integration/test_cron.py`, + `test_research_cron.py` and `test_chat_sensor_wiring.py` are not in §5's list. §5 anticipated the + repair class but enumerated three files where the real set is six. The PR body says so. +- **The daemon can read `pmset` where it runs** (stop-and-raise condition 4, checked): `/usr/bin/ + pmset` is `-rwxr-xr-x root:wheel`, `-g batt` needs no privilege, and neither plist overrides PATH, + so launchd's default PATH reaches it. Verified by reading the plists and the binary's mode, not by + executing inside the daemon — recorded as such. + +### Nothing was reached for that the plan says to stop on + +`HEAVY_TIERS` was read, never reshaped; `tests/integrity/test_shadow_isolation.py` is untouched and +green. `scheduler/presence.py` is untouched. No cancellation of an in-flight job exists anywhere in +the diff. No status field, no blessing, no `deploy`, no fixed point. + +### Next action + +The owner's review of **PR #41** (`https://github.com/ascalva/mind-palace/pull/41`). On merge, #12 +closes (the body carries the keyword unbackticked — verified after opening, since GitHub does not +parse it inside a code span) and #40 remains parked with its re-entry condition. + +### Open questions + +- **#40** (`type:direction`, `route:orchestrator`, `parked`) — the process-level clean stop is + unwired; re-entry is a fifth incident showing the idling daemon itself drains below the floor, or + someone wiring a power reading into `status` display (A1.3's parked decision). + +### Context-manifest delta + +Read beyond §2's manifest, all load-bearing: `scheduler/queue.py` (what "the ledger" is at the +supervisor's altitude — `claim` mints the only RUNNING row, `sweep_orphans` is what "clean" means to +the next run), `ops/lifecycle/runs.py` + `launcher.py:660-860` (recovery mode is the RUN ledger's, +not the queue's — the distinction Item 4's honest claim rests on), `.github/workflows/ci.yml` (the +runner is `ubuntu-latest`, which is what makes the fail-closed default a test-repair obligation), +both launchd plists (PATH and `ThrottleInterval`), `tests/integrity/test_shadow_isolation.py:90-107` +(the tripwire, confirmed untouched). `docs/findings/finding-0279.md` was NOT read — the frozen +evidence it holds is carried verbatim in Amendment A1, which was. + +```read-map +docs/design-notes/dn-supervision-and-liveness.md:668: A1.1 — the one hard pin: the power axis is its OWN predicate, never folded into blocked_tiers() +docs/design-notes/dn-supervision-and-liveness.md:684: A1.2 — the fail-closed idiom borrowed from Presence, and the floor's "clean stop, never a death" +docs/design-notes/dn-supervision-and-liveness.md:711: A1.4 — tier 5 with a tier-4 test, and the honest limit: bounds what is STARTED, not what is running +scheduler/power.py:139: the fail-closed default — an unreadable battery is discharging (A1.7's named falsifier, made a field so it is executable) +scheduler/power.py:147: the ONE named asymmetry: unknown percent does not halt every tier, or the guard against an outage becomes one +scheduler/power.py:167: below_floor — on AC is never below the floor at any percentage; the comparison is <=, and why +scheduler/power.py:106: the probe: shutil.which guard + explicit timeout + (OSError, SubprocessError) — three paths to None, none to a crash +scheduler/supervisor.py:186: power_blocked_tiers — the third sibling; HEAVY_TIERS read, never reshaped; tier claimed honestly +scheduler/supervisor.py:245: THE FLOOR, sitting BEFORE claim — that placement is the whole content of "close the ledger clean" +scheduler/supervisor.py:251: the ONE claim site: the union gains one term and nothing else changes +tests/unit/test_power.py:79: ⚑ the plan's most important test — None yields the restrictive answer, with the probe proven consulted +tests/unit/test_power.py:107: the inversion executed: flipping the default turns the unreadable case into "fine", so the default is load-bearing +tests/integration/test_supervisor.py:503: ⚑ the tier-4 composition test — deleting the union term reds exactly this and nothing else +tests/integration/test_supervisor.py:561: the floor's clean close, asserted as the next run sees it (light tier, so the shed provably does not cover it) +tests/integration/test_supervisor.py:633: the "must not spin hot" falsifier: ten ticks cost one battery reading +tests/integration/test_supervisor.py:470: the foreground gate is byte-identical under all four presence × power combinations +``` + +## Follow-through +- **Built?** Yes — sensor, predicate, composition, floor, and the repaired test surface. Five + commits, each item separately reviewable; Items 2 and 3 deliberately split so "inert" and "now + consumed" are distinct diffs. +- **Wired / delivered (or why dormant)?** Wired, and ON by default with no flag: `Supervisor.power` + is `field(default_factory=Power)`, so `launcher.py:521` and `scripts/watch.py:95` get the guard + with no edit — the ON switch is not a follow-up. What is NOT wired, deliberately and filed as + **#40**: the process-level clean stop (the launcher is out of write_scope, and a KeepAlive restart + loop would out-drain the idle hold). +- **Does a consumer use it?** Yes — the union at the one claim site is the consumer, and deleting + the term reds a test. That is the whole point of Item 3 existing separately from Item 2. +- **Track state (what remains on this track)?** `dn-supervision-and-liveness` Amendment A1 is fully + implemented by this plan; A1.3's preflight *display* of power state was never in scope and stays + parked (§11). The note's main body is untouched by this build — §2.5's non-blocking dispatch and + finding-0178's in-flight bounding remain open and are explicitly NOT this. The plan protects + bp-153's rebuild, which is the reason to land it before bp-153 Item 3 runs. +- **Opened a new track/finding?** One issue: **#40** (`type:direction`, `route:orchestrator`, + `parked`, with its re-entry condition). No new track. + +**Ready to deskcheck.** The honest demonstration is: unplug the machine and watch a synthesis-tier +job stay QUEUED while the light lanes drain, then plug in and watch it go — the code path is exactly +what the tests exercise, but a deskcheck on real hardware is what proves the *probe* reads what we +think it reads on the live daemon. + ## Pre-build notes for whoever picks this up - ⚑⚑ **The power rule gets its OWN predicate. This is the one hard pin.** Do not fold it into diff --git a/scheduler/power.py b/scheduler/power.py new file mode 100644 index 00000000..88279475 --- /dev/null +++ b/scheduler/power.py @@ -0,0 +1,184 @@ +"""Power-state detection — the scheduler's SECOND resource axis (`dn-supervision-and-liveness` +Amendment A1; issue #12). + +§2.7 gave the supervisor exactly one resource to refuse on: **memory** (non-negotiable #8, "the +scheduler refuses breaching work"). That principle was never memory-specific — it was simply never +given a second axis. The machine's remaining energy is a schedulable resource too, and work that +would spend the last of it is breaching work. Three measured emergencies are the warrant: Jul 24 +(drained to 1% during a deploy night — the embedder starved under critical-battery throttle, +`code_sync` wedged, and the daemon died unwitnessed and stayed dead three days), Jul 28 (100%->8% +in 2h40m, caught at the wire), and Aug 1 (fatal — the machine died mid-run and run #39 came up in +recovery). The battery hardware is healthy throughout: Condition Normal, 95% maximum capacity, 128 +cycles, re-measured 2026-08-05. **The drain is load, not degradation** — the scheduler was the +defect. + +This module is deliberately shaped like `scheduler/presence.py` rather than invented: an injectable +probe (so the gate is testable with no hardware, and a non-macOS worker can supply its own), a +threshold, and — the idiom that matters — a fail-CLOSED default. `Presence` reads +`assume_present_when_unknown = True`; this reads `assume_discharging_when_unknown = True`. + +⚑ **Fail closed, and why that is the load-bearing line.** `None` is an ORDINARY state here, not a +theoretical one: `pmset` may be absent (`shutil.which`), the exec may fail or hang (hence an +explicit `timeout=`), and the output may be unreadable — all three funnel into `None`, and so does +a probe that raises. The failure mode being designed against is *the machine dying*; a sensor that +failed OPEN would re-create it precisely when the system is least healthy. So an unreadable battery +is treated as discharging. + +What this module does NOT decide, deliberately: **which** tiers are shed (that is +`supervisor.HEAVY_TIERS` — read, never reshaped, so there is one shed vocabulary and not two) and +**whether** a job is refused (that is `Supervisor.power_blocked_tiers`, one of three sibling +predicates). One sensor feeding one predicate is A1.5's constraint, carried from this note's §1.2 +non-goal: N ad-hoc detectors, each with its own falsifier and its own rot, is the failure mode the +note exists to avoid. +""" + +from __future__ import annotations + +import re +import shutil +import subprocess +from collections.abc import Callable +from dataclasses import dataclass + +# The charge at which the answer stops being "shed the heavy lanes" and becomes "start nothing at +# all" (Amendment A1.2, ~20%). A lower floor buys runway and was rejected: the margin exists to +# close cleanly, and Jul 28 fell 100%->8% in 2h40m, so the tail is fast. +DEFAULT_FLOOR_PERCENT = 20.0 + + +@dataclass(frozen=True) +class PowerState: + """One sampled reading of the physical world. `percent` is optional because a host can answer + the source question ("am I on mains?") while having no battery to report at all — a desktop, a + machine with the battery removed. The two facts are separable, so they are separate fields.""" + + discharging: bool + percent: float | None = None + + +PowerProbe = Callable[[], "PowerState | None"] + +_SOURCE_PREFIX = "now drawing from" +_AC = "'ac power'" +_BATTERY = "'battery power'" +_PERCENT = re.compile(r"(\d{1,3}(?:\.\d+)?)%") + + +def parse_pmset(text: str) -> PowerState | None: + """Read `pmset -g batt` output. `None` when it cannot be read — which the caller turns into the + restrictive answer, so an unrecognized format degrades to "discharging" rather than to "fine". + + The two shapes this parses, both real (the first captured verbatim from the live machine on + 2026-08-05, the second is the discharging form the Jul 28 sampler logged):: + + Now drawing from 'AC Power' + -InternalBattery-0 (id=23068771)\t100%; charged; 0:00 remaining present: true + + Now drawing from 'Battery Power' + -InternalBattery-0 (id=23068771)\t8%; discharging; 0:21 remaining present: true + + The `Now drawing from` line is authoritative and is read first: it answers the source question + directly, whereas the per-battery status word is a vocabulary (`charged`, `charging`, + `discharging`, `finishing charge`, `AC attached`) that grows with the OS. The word is a + FALLBACK for output that lacks the source line, never an override of it. + """ + source: bool | None = None + saw_discharging_word = False + percent: float | None = None + for raw in text.splitlines(): + line = raw.strip().lower() + if source is None and line.startswith(_SOURCE_PREFIX): + if _BATTERY in line: + source = True + elif _AC in line: + source = False + if "discharging" in line: + saw_discharging_word = True + if percent is None: + found = _PERCENT.search(line) + if found is not None: + percent = float(found.group(1)) + if source is None and not saw_discharging_word: + return None # unreadable — the caller's fail-closed default takes over + return PowerState(discharging=source if source is not None else True, percent=percent) + + +def macos_power_state() -> PowerState | None: + """The machine's power state via `pmset -g batt`. `None` if unavailable (e.g. not macOS, the + exec failed or hung, or the output was unreadable). + + Three habits are carried deliberately from `presence.macos_idle_seconds`, and each one is a + path to `None` rather than to a crash or a stall: the `shutil.which` guard (an absent tool is + not an exception), the explicit `timeout=` (a hung probe must not stall dispatch — this runs on + the supervisor's own thread, inside `tick`), and the `(OSError, SubprocessError)` catch. + """ + if shutil.which("pmset") is None: + return None + try: + out = subprocess.run( + ["pmset", "-g", "batt"], capture_output=True, text=True, timeout=5 + ).stdout + except (OSError, subprocess.SubprocessError): + return None + return parse_pmset(out) + + +@dataclass +class Power: + """The power sensor, in the `Presence` mould: an injectable probe, a floor, and a fail-closed + default. Injectability is what makes the gate testable with no hardware — every test in + `tests/unit/test_power.py` injects, so no `pmset` subprocess runs there, and the real probe is + never invoked at import time or by construction (only by a call).""" + + power_probe: PowerProbe = macos_power_state + floor_percent: float = DEFAULT_FLOOR_PERCENT + # ⚑ THE FAIL-CLOSED DEFAULT — the module docstring's load-bearing line, and the amendment's + # named falsifier (A1.7: "if an unreadable/absent `pmset` yields 'not discharging' and work + # dispatches, the design is inverted"). Exposed as a field, like `Presence`'s, so the inversion + # is *executable* in a test rather than merely asserted in prose. + assume_discharging_when_unknown: bool = True + # ⚑ The ONE asymmetry in that posture, named rather than accidental. An unreadable battery is + # treated as DISCHARGING above, and the cost of that is bounded: the heavy lanes are shed, + # which is the "safe but useless" degradation the plan's §10 already accepts. An unreadable + # battery treated as BELOW THE FLOOR is a different animal — the floor refuses EVERY tier, so a + # host with no battery to read (a desktop, CI, any non-macOS worker) would dispatch nothing, + # ever, and the guard against an outage would BE an outage. So an unknown percentage does not + # halt by default; the knob exists, defaulted off, so the decision is visible and reversible. + halt_when_percent_unknown: bool = False + + def state(self) -> PowerState | None: + """The current reading, or `None` when the battery is unreadable. **Never raises**: a probe + that throws is a `None`, because an exception escaping into `tick` would be a new crash + path in the scheduler — a guard that can take down the loop it protects is not a guard.""" + try: + return self.power_probe() + except Exception: # noqa: BLE001 — every probe failure is the same fact: unreadable + return None + + def discharging(self) -> bool: + """True if the machine is running on its battery (so the heavy lanes are shed). An + unreadable battery returns the restrictive answer — see `assume_discharging_when_unknown`. + """ + state = self.state() + if state is None: + return self.assume_discharging_when_unknown + return state.discharging + + def below_floor(self) -> bool: + """True when the remaining charge has reached the floor WHILE discharging — the point at + which the supervisor starts nothing at all. + + Two boundaries stated exactly. **On AC is never below the floor**, at any percentage: the + floor's whole content is "hold for AC", so a machine that is already charging is recovering + and there is nothing to hold for. And the comparison is `<=`, not `<` — reaching the floor + counts as reaching it, because the margin exists to close down cleanly and spending it is + not one of the options. + """ + state = self.state() + if state is None: + return self.assume_discharging_when_unknown and self.halt_when_percent_unknown + if not state.discharging: + return False + if state.percent is None: + return self.halt_when_percent_unknown + return state.percent <= self.floor_percent diff --git a/scheduler/supervisor.py b/scheduler/supervisor.py index faa8003d..ed69808e 100644 --- a/scheduler/supervisor.py +++ b/scheduler/supervisor.py @@ -1,8 +1,12 @@ """The supervisor — one loop owns the queue and the worker slot (BUILD-SPEC §13; roadmap §7). Cooperative, job-boundary scheduling: + 0. refuse to start anything at all while the battery is below the floor (the POWER axis, + `dn-supervision-and-liveness` Amendment A1) — the refusal sits ahead of the claim, so no + RUNNING row is minted for a machine that may not survive to close it; 1. claim the next eligible job (priority; swap-avoidance within a priority band; heavy - tiers gated while the owner is present — the foreground check); + tiers gated while the owner is present — the foreground check — and shed while the machine + is on battery — the power check); 2. make its (tier, window) resident via the two-slot loader, which refuses any load that would breach the RAM ceiling (Invariant 8) — such a job is deferred, not crashed; 3. dispatch it, in ONE OF TWO MODES (see below), counting *worker* swaps @@ -45,6 +49,7 @@ from core.models import MemoryCeilingError from core.models.loader import TwoSlotLoader from core.stores.telemetry import TelemetryWriter +from scheduler.power import Power from scheduler.presence import Presence from scheduler.queue import RUNNING, Job, JobQueue from scheduler.worker import ( @@ -92,6 +97,13 @@ class Supervisor: loader: TwoSlotLoader handlers: dict[str, Handler] presence: Presence = field(default_factory=Presence) + # The POWER axis (`dn-supervision-and-liveness` Amendment A1). Defaulted exactly like + # `presence`, so every existing construction site — `launcher.py:521`, `scripts/watch.py:95`, + # the test sites — gets the guard with no edit; a resource refusal wired only where someone + # remembered to pass it is a refusal the daemon does not have. Its default probe FAILS CLOSED + # (an unreadable battery reads as discharging), which is why tests that dispatch a heavy tier + # inject an on-AC sensor rather than inheriting the host's real battery. + power: Power = field(default_factory=Power) telemetry: TelemetryWriter | None = None secrets: SecretsBackend | None = None # vault-runtime-auth.md; Phase 5 wires per-job use warm: bool = True # tests pass warm=False (no Ollama calls) @@ -131,7 +143,11 @@ def blocked_tiers(self) -> frozenset[str]: """THE FOREGROUND GATE, and nothing else. Deliberately not extended with the single-model-in-flight rule (bp-110 §7 Item 4's invariant: "the foreground gate keeps its meaning and is not overloaded") — two different reasons to refuse a tier, conflated into - one predicate, is how a reader later cannot tell which rule refused a job.""" + one predicate, is how a reader later cannot tell which rule refused a job. + + The siblings, so all THREE are findable from any one of them: `model_blocked_tiers` ("is a + model already out?") and `power_blocked_tiers` ("is there energy?", Amendment A1). Three + predicates, three questions, composed only by union at the ONE claim site in `tick`.""" return HEAVY_TIERS if self.presence.foreground_active() else frozenset() def model_blocked_tiers(self) -> frozenset[str]: @@ -171,10 +187,72 @@ def model_blocked_tiers(self) -> frozenset[str]: if m.tier not in (in_flight_tier, self._pinned_tier) ) + def power_blocked_tiers(self) -> frozenset[str]: + """THE POWER AXIS, and nothing else (`dn-supervision-and-liveness` Amendment A1; issue + #12): on battery, shed the heavy lanes. + + ⚑ **The third sibling, deliberately NOT folded into `blocked_tiers()`** — the amendment's + one load-bearing pin, for exactly the reason that method's docstring already gives about + the model rule. A power refusal and a presence refusal answer different questions ("is + there energy?" vs "is the owner here?"), and a reader who cannot tell which rule refused a + job cannot fix the one that is wrong. Composition happens only at the ONE claim site. + + `HEAVY_TIERS` is READ here, never reshaped: the shed vocabulary is the existing one, so + there is a single answer to "which lanes are heavy?" rather than two that can drift (A1's + parked selector decision — `load_key` was rejected as the default because it introduces a + second, finer vocabulary whose interaction with this set nobody has designed). + + Tier accounting, stated honestly per A1.4: a dispatch guard — **tier 5 with a tier-4 + test**, deliberately identical to what §2.7 claims for the memory ceiling. Power is a + sampled reading of the physical world, so no value can be made to not inhabit "the battery + is low": tier 1 is unreachable here and claiming it would be the overclaim §0's ladder + names as *the* foot-gun. What the tier-4 test buys is `tests/integration/test_supervisor.py` + proving the union in `tick` actually contains this term, and that the probe's `None` path + fails closed — a predicate nobody calls is the finding-0187 shape (deleting bp-105's sweep + call left 85/85 green). + + ⚑ **The honest limit, recorded rather than hidden (A1.4):** this bounds what is STARTED, + never what is already running. Jul 24's `code_backfill` was in flight when the throttle + hit, so this would not have prevented that emergency outright. In-flight energy bounding + needs the job-timeout machinery (finding-0178) and is not designed here; nothing in this + path ever kills a running job. + """ + return HEAVY_TIERS if self.power.discharging() else frozenset() + def tick(self) -> bool: - """Dispatch at most one job. Returns False when nothing is runnable right now.""" + """Dispatch at most one job. Returns False when nothing is runnable right now. + + ⚑ **THE ONE CLAIM SITE.** All three refusal predicates — presence, single-model, power — + compose HERE, by union, and nowhere else (`model_blocked_tiers`'s pin: "Enforced at the ONE + claim site, via `claim`'s existing `blocked_tiers` — no new queue API"). Each stays + separately readable so a reader can still tell which rule refused a job; the union is the + only place they are indistinguishable, and it is one line long.""" + # ⚑ THE POWER FLOOR (Amendment A1.2), and note it sits BEFORE `claim`. Below the floor the + # answer stops being "shed the heavy lanes" and becomes "start nothing at all": every tier + # is refused, including the light ones the shed above leaves alone. Refusing before the + # claim is what closes the ledger clean — no RUNNING row is minted, so nothing is left for + # a machine that may not survive to close it, and the orphan sweep of whatever run follows + # has nothing to reclaim. Claiming here would reproduce the Aug 1 shape exactly: the + # machine died mid-run and run #39 came up in recovery. + # + # ⚑ The hold for AC is the ABSENCE of dispatch, not a wait loop. Returning False ends the + # drain, and the launcher's existing conditional sleep is the duty cycle + # (`ops/lifecycle/launcher.py`: "sleep only when the drain came back idle"), so the next + # tick re-reads the battery and dispatch resumes by itself once mains are back. An + # in-process sleep/wait here was the rejected alternative (A1's parked hold-for-AC + # decision): it would hold the supervisor lock while doing nothing and would itself drain + # the battery it is protecting. + # + # ⚑ And nothing is killed. This bounds what is STARTED; a job already in flight runs to its + # own completion or checkpoint (A1.4's honest limit — in-flight energy bounding is + # finding-0178's job-timeout machinery, not this). + if self.power.below_floor(): + return False + job = self.queue.claim(loaded_key=self._worker_key, - blocked_tiers=self.blocked_tiers() | self.model_blocked_tiers()) + blocked_tiers=(self.blocked_tiers() + | self.model_blocked_tiers() + | self.power_blocked_tiers())) if job is None: return False @@ -262,7 +340,12 @@ def _dispatch_to_worker(self, job: Job) -> bool: def run(self, *, max_ticks: int | None = None) -> int: """Drain the queue cooperatively. Returns the number of jobs dispatched. Stops when - nothing is runnable (e.g. only heavy jobs remain while the owner is present).""" + nothing is runnable (e.g. only heavy jobs remain while the owner is present, or while the + machine is on battery — and, below the power floor, when nothing at all may start). + + Below the floor this returns 0 on the FIRST tick and returns control; it never loops or + sleeps waiting for mains. That is the hold, and it is deliberately the caller's duty cycle + rather than one invented here (Amendment A1's parked hold-for-AC decision).""" n = 0 while max_ticks is None or n < max_ticks: if not self.tick(): diff --git a/tests/e2e/test_scheduler_live.py b/tests/e2e/test_scheduler_live.py index 478e98ab..9d613341 100644 --- a/tests/e2e/test_scheduler_live.py +++ b/tests/e2e/test_scheduler_live.py @@ -9,6 +9,7 @@ from scheduler.presence import Presence from scheduler.queue import DONE, JobQueue from scheduler.supervisor import Supervisor +from tests.fixtures.power import on_ac pytestmark = pytest.mark.live @@ -43,6 +44,9 @@ def handler(_job): loader=server.loader, handlers={"ping": handler}, presence=Presence(idle_probe=lambda: 10_000.0), # idle => nothing gated + # bp-154: on mains, so this live gate measures Ollama and the loader — not whatever the + # developer's battery happens to be doing while the suite runs. + power=on_ac(), ) j = sup.queue.enqueue("ping", "router", cfg.pinned_model.num_ctx) assert sup.run() == 1 diff --git a/tests/fixtures/power.py b/tests/fixtures/power.py new file mode 100644 index 00000000..beb5a4b4 --- /dev/null +++ b/tests/fixtures/power.py @@ -0,0 +1,35 @@ +"""Injected power sensors for the scheduler tests (`dn-supervision-and-liveness` Amendment A1). + +`Supervisor.power` defaults to a sensor that FAILS CLOSED: a host that cannot read `pmset` — CI, +any non-macOS worker — reads as discharging, and a host that can read it reports whatever the +developer's battery happens to be doing right now. Either way a default-constructed supervisor in a +test would decide heavy-tier dispatch from the machine the suite runs on. So every test that +constructs a real `Supervisor` injects one of these instead, exactly as it already injects +`Presence(idle_probe=...)` rather than reading the host's real HID idle time. + +⚑ One rule about which one to inject. `on_ac()` restores the intended subject of a test that was +never about power (the foreground gate, the ceiling, the dispatch seam). Injecting it into a test +that is *meant* to exercise a power refusal would hide the feature rather than accommodate it — +bp-154 Item 5's named falsifier. +""" + +from __future__ import annotations + +from scheduler.power import Power, PowerState + + +def on_ac(percent: float = 100.0) -> Power: + """Plugged in — the power axis refuses nothing.""" + return Power(power_probe=lambda: PowerState(discharging=False, percent=percent)) + + +def on_battery(percent: float = 55.0) -> Power: + """Running on the battery. The default is deliberately well above the floor, so a test using it + exercises the discharging shed and NOT the floor; pass a low percentage to reach the floor.""" + return Power(power_probe=lambda: PowerState(discharging=True, percent=percent)) + + +def unreadable() -> Power: + """A battery that cannot be read at all (no `pmset`, a failed exec, unparseable output). The + fail-closed default makes this discharging.""" + return Power(power_probe=lambda: None) diff --git a/tests/integration/test_chat_sensor_wiring.py b/tests/integration/test_chat_sensor_wiring.py index aa44a18a..0e4de25f 100644 --- a/tests/integration/test_chat_sensor_wiring.py +++ b/tests/integration/test_chat_sensor_wiring.py @@ -38,6 +38,7 @@ from scheduler.queue import DONE, JobQueue from scheduler.router import Router from scheduler.supervisor import Supervisor +from tests.fixtures.power import on_ac from tests.unit.test_loader_reconcile import loader_for @@ -72,6 +73,7 @@ def test_daemon_runs_a_chat_sync_job_and_the_store_gains_rows(tmp_path: Path) -> queue=queue, loader=_loader(cfg), handlers={CHAT_SYNC_KIND: chat_sync_handler(sensor)}, presence=Presence(idle_probe=lambda: 10_000.0), # owner idle → nothing gated + power=on_ac(), # bp-154: on mains → nothing shed either warm=False, ) sup.loader.ensure_pinned(warm=False) # pinned tier resident, no Ollama call diff --git a/tests/integration/test_cron.py b/tests/integration/test_cron.py index 6ce1c88e..482c9afc 100644 --- a/tests/integration/test_cron.py +++ b/tests/integration/test_cron.py @@ -23,6 +23,7 @@ from scheduler.queue import QUEUED, Job, JobQueue from scheduler.router import Router from scheduler.supervisor import Supervisor +from tests.fixtures.power import on_ac from tests.unit.test_loader_reconcile import loader_for @@ -66,9 +67,13 @@ def test_cron_jobs_are_gated_during_foreground_then_run_in_a_trough(tmp_path): queue = JobQueue(tmp_path / "q.db") def make_supervisor(active): + # bp-154: an on-AC power sensor, injected for the same reason `_present` is — this test's + # subject is the FOREGROUND gate, and `Supervisor.power` defaults to a sensor that fails + # closed, so a default construction would decide these synthesis-tier dispatches from the + # host's battery (or, on CI, from the absence of `pmset`). return Supervisor(queue=queue, loader=_loader(cfg), handlers=cron_handlers(dreamer, curator), - presence=_present(active), warm=False) + presence=_present(active), power=on_ac(), warm=False) d = enqueue_dream(queue, router) c = enqueue_curate(queue, router) diff --git a/tests/integration/test_research_cron.py b/tests/integration/test_research_cron.py index 185eff5e..86b87d66 100644 --- a/tests/integration/test_research_cron.py +++ b/tests/integration/test_research_cron.py @@ -25,6 +25,7 @@ from scheduler.research import RESEARCH_KIND from scheduler.router import Router from scheduler.supervisor import Supervisor +from tests.fixtures.power import on_ac from tests.unit.test_loader_reconcile import loader_for @@ -105,9 +106,12 @@ def test_research_is_gated_during_foreground_then_runs_in_a_trough(tmp_path, mon handler = research_handler(_airlock(airlock), _emb(), store=_store()) def make_supervisor(active): + # bp-154: an on-AC power sensor, injected for the same reason `_present` is — this test's + # subject is the FOREGROUND gate on a synthesis-tier job, and `Supervisor.power` defaults + # to a sensor that fails closed (on CI, where `pmset` is absent, that means discharging). return Supervisor(queue=queue, loader=_loader(cfg), handlers={RESEARCH_KIND: handler}, - presence=_present(active), warm=False) + presence=_present(active), power=on_ac(), warm=False) job = enqueue_research(queue, router, criteria) # The enqueued payload is de-identified — no raw query text crosses into the queue (Inv 11). diff --git a/tests/integration/test_supervisor.py b/tests/integration/test_supervisor.py index 0978b370..78936536 100644 --- a/tests/integration/test_supervisor.py +++ b/tests/integration/test_supervisor.py @@ -10,15 +10,18 @@ "no behaviour change at landing" claim is false (bp-110 §7 Item 3's falsifier).""" import dataclasses +import inspect from collections.abc import Callable import pytest from config.loader import load_config +from scheduler.power import Power, PowerState from scheduler.presence import Presence -from scheduler.queue import DEFERRED, DONE, FAILED, QUEUED, JobQueue -from scheduler.supervisor import SUBPROCESS, Supervisor +from scheduler.queue import DEFERRED, DONE, FAILED, QUEUED, RUNNING, JobQueue +from scheduler.supervisor import HEAVY_TIERS, SUBPROCESS, Supervisor from scheduler.worker import SELFTEST_ANSWER_KIND, Batch +from tests.fixtures.power import on_ac, on_battery, unreadable from tests.fixtures.secrets import fake_vault from tests.unit.test_loader_reconcile import loader_for @@ -34,12 +37,20 @@ def _present(active: bool) -> Presence: return Presence(idle_probe=lambda: 0.0 if active else 10_000.0) -def _supervisor(tmp_path, handlers, *, active=False, loader=None, secrets=None): +def _supervisor(tmp_path, handlers, *, active=False, loader=None, secrets=None, power=None, + queue=None): return Supervisor( - queue=JobQueue(tmp_path / "q.db"), + queue=queue or JobQueue(tmp_path / "q.db"), loader=loader or _loader(), handlers=handlers, presence=_present(active), + # bp-154: `Supervisor.power` fails CLOSED by default (an unreadable battery reads as + # discharging), so a default-constructed supervisor here would decide heavy-tier dispatch + # from whatever the host's battery is doing — or, on CI, from the absence of `pmset`. + # Injecting an on-AC sensor restores each test's intended subject, exactly as `_present` + # already does for HID idle time. The power tests at the bottom of the file inject their + # own; injecting `on_ac()` into one of THOSE would hide the feature (Item 5's falsifier). + power=power or on_ac(), warm=False, secrets=secrets, ) @@ -90,7 +101,8 @@ def test_ceiling_breach_defers_job(tmp_path): ld = _loader(cfg) ld.ensure_pinned(warm=False) # 2.7 GB of a 5 GB budget sup = Supervisor(queue=JobQueue(tmp_path / "q.db"), loader=ld, - handlers={"k": lambda j: None}, presence=_present(False), warm=False) + handlers={"k": lambda j: None}, presence=_present(False), power=on_ac(), + warm=False) j = sup.queue.enqueue("k", "synthesis", 32768) # 2.7 + 17 > 5 -> refused sup.run() deferred = sup.queue.get(j.id) @@ -283,7 +295,7 @@ def test_the_ceiling_gate_still_refuses_BEFORE_any_worker_is_spawned(tmp_path): ld = _loader(cfg) ld.ensure_pinned(warm=False) sup = Supervisor(queue=JobQueue(tmp_path / "q.db"), loader=ld, handlers={}, - presence=_present(False), warm=False) + presence=_present(False), power=on_ac(), warm=False) sup.worker_mode = SUBPROCESS sup.compute[SELFTEST_ANSWER_KIND] = ( lambda j, c: (_ for _ in ()).throw(AssertionError("spawned past the ceiling gate")), @@ -417,3 +429,223 @@ def test_a_crashed_worker_does_not_strand_the_model_gate_closed(tmp_path): sup.run() assert sup._in_flight_key is None # released despite the worker's death assert sup.model_blocked_tiers() == frozenset() + + +# ============================================================================================== +# bp-154 Item 2 — the power axis, as its OWN predicate (dn-supervision-and-liveness A1) +# ============================================================================================== + + +def test_the_heavy_lanes_are_shed_while_discharging(tmp_path): + """⚑ Item 2's acceptance. On battery, `power_blocked_tiers()` sheds exactly the existing heavy + set — READ, never reshaped, so there is one shed vocabulary rather than two (A1's parked + selector decision).""" + sup = _supervisor(tmp_path, {}, power=on_battery(55.0)) + # Non-vacuity, twice over: the shed set is not empty (an empty `HEAVY_TIERS` would make every + # assertion below trivially true), and 55% is well above the floor, so what refuses here is the + # DISCHARGING rule and not the floor. + assert HEAVY_TIERS == frozenset({"synthesis", "stretch"}) + assert sup.power.below_floor() is False + assert sup.power_blocked_tiers() == HEAVY_TIERS + + +def test_on_AC_the_power_axis_refuses_nothing(tmp_path): + """The rule is not a permanent shed dressed up as a safety property: plugged in, it says + nothing at all.""" + sup = _supervisor(tmp_path, {}, power=on_ac()) + assert sup.power_blocked_tiers() == frozenset() + + +def test_an_unreadable_battery_sheds_the_heavy_lanes(tmp_path): + """⚑ Fail closed, at the supervisor's altitude rather than only the sensor's (A1.7's named + falsifier: "the sensor fails open"). A host that cannot read `pmset` at all — CI, a non-macOS + worker, a failed exec — refuses heavy work rather than dispatching it blind.""" + sup = _supervisor(tmp_path, {}, power=unreadable()) + assert sup.power.state() is None # precondition: genuinely unreadable + assert sup.power_blocked_tiers() == HEAVY_TIERS + + +@pytest.mark.parametrize("discharging", [True, False]) +@pytest.mark.parametrize("active", [True, False]) +def test_the_foreground_gate_is_byte_identical_under_every_power_state(tmp_path, active, + discharging): + """⚑ Item 2's invariant, and A1.1's one load-bearing pin. `blocked_tiers()` is THE FOREGROUND + GATE and nothing else: its answer is a function of presence alone, unchanged by the power axis + under all four combinations. Two different reasons to refuse a tier, conflated into one + predicate, is how a reader later cannot tell which rule refused a job.""" + power = on_battery(55.0) if discharging else on_ac() + sup = _supervisor(tmp_path, {}, active=active, power=power) + assert sup.blocked_tiers() == (HEAVY_TIERS if active else frozenset()) + # ... and the power predicate is independently correct at the same time, so the two rules are + # provably separate answers rather than one answer read twice. + assert sup.power_blocked_tiers() == (HEAVY_TIERS if discharging else frozenset()) + + +def test_the_power_rule_is_not_folded_into_the_foreground_gate(tmp_path): + """The pin, read off the SOURCE — behaviour alone cannot distinguish "a separate predicate" + from "one predicate that happens to agree today", and it is the shape, not the answer, that + A1.1 fixes. Each of the three predicates reads its own sensor and no other's.""" + foreground = inspect.getsource(Supervisor.blocked_tiers).split('"""')[-1] + power = inspect.getsource(Supervisor.power_blocked_tiers).split('"""')[-1] + # Non-vacuity: the token IS findable by this search where it legitimately appears, so its + # absence from `blocked_tiers`'s body is evidence and not an artifact of a search that matches + # nothing anywhere. + assert "self.power." in inspect.getsource(Supervisor) + assert "self.power." not in foreground and "self.presence." in foreground + assert "self.presence." not in power and "self.power." in power + + +# ============================================================================================== +# bp-154 Item 3 — composed at the ONE claim site (the tier-4 test) +# ============================================================================================== + + +def test_a_heavy_job_is_NOT_claimed_while_discharging_and_IS_once_back_on_AC(tmp_path): + """⚑ Item 3's acceptance, and the whole reason Items 2 and 3 are separate. A predicate that is + green in isolation but never composed into `claim` is decorative — the finding-0187 shape + (deleting bp-105's sweep call left 85/85 green). This test fails if the + `| self.power_blocked_tiers()` term is deleted from the union in `tick`, which is the property + that makes the guard real rather than present. + + The SAME queue is drained twice — once on battery, once on AC — so the only difference between + "left QUEUED" and "ran" is the power reading.""" + ran: list[str] = [] + queue = JobQueue(tmp_path / "q.db") + sup = _supervisor(tmp_path, {"k": lambda j: ran.append(j.tier)}, active=False, + power=on_battery(55.0), queue=queue) + sup.loader.ensure_pinned(warm=False) + sup.queue.enqueue("k", "routine", 16384) + syn = sup.queue.enqueue("k", "synthesis", 32768) + + # ⚑ Non-vacuity: NEITHER of the other two refusal rules is armed, and the floor is not reached. + # Without these four assertions the test would pass just as well if the foreground gate (or the + # model rule, or the floor) were what left the synthesis job QUEUED — i.e. it would not be a + # test of the power term at all. + assert sup.blocked_tiers() == frozenset() # the owner is idle: the gate is open + assert sup.model_blocked_tiers() == frozenset() # no worker is out + assert sup.power.below_floor() is False # 55% — the floor is not what refuses + assert sup.power_blocked_tiers() == HEAVY_TIERS # this rule, and only this rule, refuses + + assert sup.run() == 1 + assert ran == ["routine"] # the light lane still drains + assert sup.queue.get(syn.id).state == QUEUED # the heavy one waits for mains + + # Plugged back in: the same queue, the same job, dispatched. The rule defers, never drops. + on_mains = _supervisor(tmp_path, {"k": lambda j: ran.append(j.tier)}, active=False, + power=on_ac(), queue=queue) + on_mains.loader.ensure_pinned(warm=False) + assert on_mains.run() == 1 + assert ran == ["routine", "synthesis"] + assert queue.get(syn.id).state == DONE + + +def test_the_power_term_composes_WITH_the_foreground_gate_rather_than_replacing_it(tmp_path): + """The union is a union: adding a term must not weaken the terms already there. On AC with the + owner present, the foreground gate alone must still refuse the heavy lane — the regression a + rewritten claim line could silently introduce.""" + ran: list[str] = [] + sup = _supervisor(tmp_path, {"k": lambda j: ran.append(j.tier)}, active=True, power=on_ac()) + sup.loader.ensure_pinned(warm=False) + sup.queue.enqueue("k", "routine", 16384) + syn = sup.queue.enqueue("k", "synthesis", 32768) + assert sup.power_blocked_tiers() == frozenset() # precondition: power refuses nothing here + assert sup.run() == 1 + assert ran == ["routine"] and sup.queue.get(syn.id).state == QUEUED + + +# ============================================================================================== +# bp-154 Item 4 — the floor: close the ledger clean and hold for AC +# ============================================================================================== + + +def test_below_the_floor_nothing_starts_and_the_ledger_is_left_CLEAN(tmp_path): + """⚑ Item 4's acceptance. Below the floor the answer is not "shed the heavy lanes" but "start + nothing at all", and the refusal happens BEFORE `claim` — so no RUNNING row is minted for a + machine that may not survive to close it. + + The falsifier is the Aug 1 shape: a stop that leaves the queue as a crash would (a stale + RUNNING row) reproduces the recovery run this exists to prevent. So the test asserts the ledger + is closed CLEAN — a following run's orphan sweep finds nothing to reclaim and nothing to + strand-fail.""" + ran: list[int] = [] + sup = _supervisor(tmp_path, {"k": lambda j: ran.append(j.id)}, active=False, + power=on_battery(5.0)) + sup.loader.ensure_pinned(warm=False) + job = sup.queue.enqueue("k", "routine", 16384) + + # ⚑ Non-vacuity: the job is on a LIGHT tier, so the discharging shed does not cover it. Without + # this the test would pass against the Item 2/3 shed alone and prove nothing about the floor. + assert job.tier not in HEAVY_TIERS + assert sup.power_blocked_tiers() == HEAVY_TIERS # the shed is armed but does not reach here + assert sup.power.below_floor() is True # the floor is what refuses + + assert sup.run() == 0 + assert ran == [] + assert sup.queue.get(job.id).state == QUEUED # waiting, not running, not failed + assert sup.queue.list(RUNNING) == [] # ⚑ nothing was claimed and abandoned + + # THE CLEAN CLOSE, asserted as the next run would experience it: a fresh run's sweep has + # nothing to reclaim or fail, which is what "the following start is not a recovery run" means + # at the queue's altitude. + sweep = sup.queue.sweep_orphans(9999) + assert (sweep.requeued, sweep.failed, sweep.total) == ((), (), 0) + + +def test_the_floor_does_not_strand_work_that_already_finished(tmp_path): + """A job that ran BEFORE the battery reached the floor must be left DONE, not stranded — the + supervisor stops starting work, it does not abandon work it already landed.""" + ran: list[int] = [] + queue = JobQueue(tmp_path / "q.db") + on_mains = _supervisor(tmp_path, {"k": lambda j: ran.append(j.id)}, power=on_ac(), queue=queue) + on_mains.loader.ensure_pinned(warm=False) + first = queue.enqueue("k", "routine", 16384) + assert on_mains.run() == 1 and ran == [first.id] # precondition: it really did run + + second = queue.enqueue("k", "routine", 16384) + drained = _supervisor(tmp_path, {"k": lambda j: ran.append(j.id)}, power=on_battery(3.0), + queue=queue) + drained.loader.ensure_pinned(warm=False) + assert drained.run() == 0 + assert queue.get(first.id).state == DONE # the finished job is untouched + assert queue.get(second.id).state == QUEUED # the new one simply never started + assert queue.list(RUNNING) == [] + + +def test_mains_returning_resumes_dispatch_without_intervention(tmp_path): + """The hold is a hold, not a wedge: the same queue drains as soon as a later reading says AC. + Under launchd KeepAlive a restart re-evaluates the same way — this is the "come back, look + again" shape, with no state to reset.""" + ran: list[int] = [] + queue = JobQueue(tmp_path / "q.db") + job = queue.enqueue("k", "routine", 16384) + + held = _supervisor(tmp_path, {"k": lambda j: ran.append(j.id)}, power=on_battery(5.0), + queue=queue) + held.loader.ensure_pinned(warm=False) + assert held.run() == 0 and queue.get(job.id).state == QUEUED + + resumed = _supervisor(tmp_path, {"k": lambda j: ran.append(j.id)}, power=on_ac(), queue=queue) + resumed.loader.ensure_pinned(warm=False) + assert resumed.run() == 1 and ran == [job.id] + assert queue.get(job.id).state == DONE + + +def test_the_hold_does_not_spin_and_does_not_drain_what_it_protects(tmp_path): + """⚑ Item 4's second falsifier: "falsified if the hold spins hot — holding for AC must not + itself consume the battery it is protecting." + + Counting the probe is how that becomes checkable. A ten-tick drain request must cost exactly + ONE reading and return immediately: `run` breaks on the first refusal instead of looping, and + there is no in-process sleep holding the supervisor lock while doing nothing (the rejected + alternative in A1's parked hold-for-AC decision).""" + readings: list[int] = [] + + def counting_probe(): + readings.append(1) + return PowerState(discharging=True, percent=5.0) + + sup = _supervisor(tmp_path, {"k": lambda j: "ok"}, power=Power(power_probe=counting_probe)) + sup.loader.ensure_pinned(warm=False) + sup.queue.enqueue("k", "routine", 16384) + assert sup.run(max_ticks=10) == 0 + assert readings == [1], f"the floor branch sampled {len(readings)} times for one drain" diff --git a/tests/unit/test_power.py b/tests/unit/test_power.py new file mode 100644 index 00000000..d583df6e --- /dev/null +++ b/tests/unit/test_power.py @@ -0,0 +1,261 @@ +"""The power sensor (`scheduler/power.py`; `dn-supervision-and-liveness` Amendment A1, bp-154 +Item 1) — the scheduler's second resource axis, in the `Presence` mould. + +⚑ **The single most important test in this file is `test_an_unreadable_battery_is_treated_as_ +discharging`**, and its sibling for a probe that raises. Amendment A1.7's named falsifier is *"the +sensor fails open"*: if an unreadable or absent `pmset` yielded "not discharging" and work +dispatched, the design would be inverted — it would fail exactly when the machine is least healthy. +`test_the_fail_closed_default_is_what_makes_that_true` executes the inversion rather than trusting +prose, so the default is proven load-bearing instead of merely present. + +Every test here injects its probe. No `pmset` subprocess runs in this file, and the real probe is +proven not to run at import or at construction (`test_constructing_the_sensor_runs_no_subprocess`). +""" + +from __future__ import annotations + +import subprocess + +import pytest + +from scheduler import power as power_mod +from scheduler.power import ( + DEFAULT_FLOOR_PERCENT, + Power, + PowerState, + macos_power_state, + parse_pmset, +) + +# Captured verbatim from the live machine on 2026-08-05 (`pmset -g batt`, on mains, 100%). +AC_OUTPUT = ( + "Now drawing from 'AC Power'\n" + " -InternalBattery-0 (id=23068771)\t100%; charged; 0:00 remaining present: true\n" +) +# The discharging shape, as the Jul 28 sampler logged it on the way from 100% to 8% in 2h40m. +BATTERY_OUTPUT = ( + "Now drawing from 'Battery Power'\n" + " -InternalBattery-0 (id=23068771)\t8%; discharging; 0:21 remaining present: true\n" +) + + +def _probe(state: PowerState | None) -> tuple[power_mod.PowerProbe, list[int]]: + """A probe returning `state`, plus the call log that makes a claim about it non-vacuous: a test + asserting the fail-closed answer must also show the probe was actually consulted, or it would + pass just as happily against a sensor that never looked.""" + calls: list[int] = [] + + def probe() -> PowerState | None: + calls.append(1) + return state + + return probe, calls + + +# --- the readings ------------------------------------------------------------------------------ + + +def test_a_discharging_probe_reports_discharging_with_its_percentage(): + probe, calls = _probe(PowerState(discharging=True, percent=55.0)) + sensor = Power(power_probe=probe) + assert sensor.discharging() is True + assert sensor.state() == PowerState(discharging=True, percent=55.0) + # Non-vacuity: 55% is deliberately well ABOVE the floor, so this reading exercises the + # discharging rule and nothing else — the floor must stay quiet here. + assert 55.0 > DEFAULT_FLOOR_PERCENT and sensor.below_floor() is False + assert calls, "the sensor answered without consulting its probe" + + +def test_an_ac_probe_reports_not_discharging(): + probe, calls = _probe(PowerState(discharging=False, percent=100.0)) + sensor = Power(power_probe=probe) + assert sensor.discharging() is False + assert calls + + +# --- ⚑ fail closed — the plan's most important case --------------------------------------------- + + +def test_an_unreadable_battery_is_treated_as_discharging(): + """⚑ Item 1's falsifier. `None` is an ORDINARY state (no `pmset`, a failed exec, unparseable + output), and it must yield the RESTRICTIVE answer.""" + probe, calls = _probe(None) + sensor = Power(power_probe=probe) + # Precondition: the default under test is the fail-closed one, not something a fixture set. + assert sensor.assume_discharging_when_unknown is True + assert sensor.state() is None + assert sensor.discharging() is True + assert calls, "the sensor answered without consulting its probe" + + +def test_a_probe_that_raises_is_treated_as_discharging_and_does_not_propagate(): + """An exception escaping into `tick` would be a new crash path in the scheduler, not a guard. + Every probe failure is the same fact — unreadable — so it lands on the same answer.""" + raised: list[str] = [] + + def exploding_probe() -> PowerState | None: + raised.append("boom") + raise OSError("pmset went away mid-read") + + sensor = Power(power_probe=exploding_probe) + assert sensor.state() is None # swallowed, not propagated + assert sensor.discharging() is True # and it lands on the restrictive answer + assert sensor.below_floor() is False # unknown percent does not halt (see below) + assert raised, "the probe never ran — this test would pass against a sensor that never looks" + + +def test_the_fail_closed_default_is_what_makes_that_true(): + """The inversion, executed. Flipping the one field turns the unreadable case into "fine" — + which is precisely A1.7's falsifier, and proves the default above is load-bearing rather than + incidental.""" + probe, _ = _probe(None) + assert Power(power_probe=probe, assume_discharging_when_unknown=False).discharging() is False + + +# --- the floor --------------------------------------------------------------------------------- + + +def test_the_floor_is_reached_only_while_discharging(): + """"Hold for AC" has no content on AC: a machine that is charging is recovering, at any + percentage. The pair is asserted at the SAME percentage, so `discharging` is provably the only + difference between the two answers.""" + low = 5.0 + assert low < DEFAULT_FLOOR_PERCENT + on_battery, _ = _probe(PowerState(discharging=True, percent=low)) + on_mains, _ = _probe(PowerState(discharging=False, percent=low)) + assert Power(power_probe=on_battery).below_floor() is True + assert Power(power_probe=on_mains).below_floor() is False + + +@pytest.mark.parametrize( + ("percent", "expected"), + [(19.9, True), (20.0, True), (20.1, False), (95.0, False)], +) +def test_the_floor_boundary_is_inclusive(percent: float, expected: bool): + """`<=`, not `<`: reaching the floor counts as reaching it, because the margin exists to close + down cleanly and spending it is not one of the options. The 20.0 row is the boundary the + comparison would silently move if it were rewritten as `<`.""" + assert DEFAULT_FLOOR_PERCENT == 20.0 + probe, _ = _probe(PowerState(discharging=True, percent=percent)) + assert Power(power_probe=probe).below_floor() is expected + + +def test_an_unknown_percentage_does_not_halt_by_default_but_the_knob_exists(): + """The one named asymmetry in the fail-closed posture. Unknown percent must not halt EVERY + tier by default — a host with no battery to read (a desktop, CI, a non-macOS worker) would then + dispatch nothing ever, and the guard against an outage would be an outage. The knob is + asserted in both positions so the default is a decision, not an accident.""" + probe, _ = _probe(PowerState(discharging=True, percent=None)) + assert Power(power_probe=probe).halt_when_percent_unknown is False + assert Power(power_probe=probe).below_floor() is False + assert Power(power_probe=probe, halt_when_percent_unknown=True).below_floor() is True + # And the same for a wholly unreadable battery: discharging (fail closed), but not halted. + blind, _ = _probe(None) + assert Power(power_probe=blind).discharging() is True + assert Power(power_probe=blind).below_floor() is False + assert Power(power_probe=blind, halt_when_percent_unknown=True).below_floor() is True + + +# --- the real probe: parsing, and the guards that funnel into `None` ----------------------------- + + +def test_parse_reads_the_two_real_pmset_shapes(): + ac = parse_pmset(AC_OUTPUT) + battery = parse_pmset(BATTERY_OUTPUT) + assert ac == PowerState(discharging=False, percent=100.0) + assert battery == PowerState(discharging=True, percent=8.0) + # Non-vacuity: the fixtures are the real formats, so a parser that only handled one of them — + # or that keyed off the status word alone ("charged" vs "discharging") — fails here. + assert "AC Power" in AC_OUTPUT and "Battery Power" in BATTERY_OUTPUT + + +def test_the_source_line_outranks_the_status_word(): + """`Now drawing from` answers the source question directly; the per-battery status word is a + vocabulary that grows with the OS (`charged`, `charging`, `finishing charge`, `AC attached`). + A machine plugged in at 30% reads `charging` — and a parser keyed off the word list would have + to know every member. The word is a fallback for output missing the source line, not an + override of it.""" + charging = ( + "Now drawing from 'AC Power'\n" + " -InternalBattery-0 (id=23068771)\t30%; charging; 1:12 remaining present: true\n" + ) + assert parse_pmset(charging) == PowerState(discharging=False, percent=30.0) + # The fallback path: no source line at all, but the word is there. + assert parse_pmset(" -InternalBattery-0\t8%; discharging; 0:21 remaining") == PowerState( + discharging=True, percent=8.0 + ) + + +def test_a_host_on_mains_with_no_battery_reports_a_state_with_no_percentage(): + """A desktop answers the source question while having no battery to report. The two facts are + separable, so they are separate fields — and `percent=None` must not be read as 0%.""" + assert parse_pmset("Now drawing from 'AC Power'\n") == PowerState( + discharging=False, percent=None + ) + + +@pytest.mark.parametrize("text", ["", "\n\n", "some other tool's output entirely"]) +def test_unreadable_output_parses_to_None(text: str): + """The third path to `None`, and the reason the fail-closed default is load-bearing: an + unrecognized format degrades to "discharging", never to "fine".""" + assert parse_pmset(text) is None + assert Power(power_probe=lambda: parse_pmset(text)).discharging() is True + + +def test_an_absent_pmset_yields_None_rather_than_an_exception(monkeypatch): + """The `shutil.which` guard, carried from `presence.macos_idle_seconds`: an absent tool is a + `None`, not a crash — which is what a non-macOS worker sees.""" + monkeypatch.setattr(power_mod.shutil, "which", lambda _name: None) + assert macos_power_state() is None + + +@pytest.mark.parametrize("boom", [OSError("no such tool"), subprocess.TimeoutExpired("pmset", 5)]) +def test_a_failed_or_hung_exec_yields_None_rather_than_an_exception(monkeypatch, boom): + """The other two carried habits: the explicit `timeout=` (a hung probe must not stall dispatch + — this runs on the supervisor's own thread inside `tick`) and the + `(OSError, SubprocessError)` catch. Both funnel to `None`.""" + monkeypatch.setattr(power_mod.shutil, "which", lambda _name: "/usr/bin/pmset") + + def explode(*_a, **_kw): + raise boom + + monkeypatch.setattr(power_mod.subprocess, "run", explode) + assert macos_power_state() is None + + +def test_the_probe_passes_an_explicit_timeout(monkeypatch): + """The timeout is asserted, not assumed: a probe without one can hang the supervisor's tick + forever, and nothing else in the suite would notice its removal.""" + seen: dict[str, object] = {} + + def fake_run(argv, **kwargs): + seen["argv"] = argv + seen["timeout"] = kwargs.get("timeout") + return subprocess.CompletedProcess(argv, 0, stdout=AC_OUTPUT, stderr="") + + monkeypatch.setattr(power_mod.shutil, "which", lambda _name: "/usr/bin/pmset") + monkeypatch.setattr(power_mod.subprocess, "run", fake_run) + assert macos_power_state() == PowerState(discharging=False, percent=100.0) + assert seen["argv"] == ["pmset", "-g", "batt"] + assert seen["timeout"] == 5 + + +def test_constructing_the_sensor_runs_no_subprocess(monkeypatch): + """Item 1's invariant: the real probe must not be invoked at import time or by construction. + The call counter is checked in BOTH directions — zero after construction, and one after a + reading — so the test cannot pass against a probe that is simply dead.""" + calls: list[list[str]] = [] + + def fake_run(argv, **_kwargs): + calls.append(argv) + return subprocess.CompletedProcess(argv, 0, stdout=AC_OUTPUT, stderr="") + + monkeypatch.setattr(power_mod.shutil, "which", lambda _name: "/usr/bin/pmset") + monkeypatch.setattr(power_mod.subprocess, "run", fake_run) + + sensor = Power() + assert sensor.power_probe is macos_power_state # the default binding, not an injected double + assert calls == [] # construction consults nothing + + assert sensor.discharging() is False # ... but a READING does + assert calls == [["pmset", "-g", "batt"]]