diff --git a/docs/superpowers/contracts/phone-lease-2026-09-20.json b/docs/superpowers/contracts/phone-lease-2026-09-20.json new file mode 100644 index 00000000..a35c444a --- /dev/null +++ b/docs/superpowers/contracts/phone-lease-2026-09-20.json @@ -0,0 +1,15 @@ +{ + "title": "One phone, one driving session at a time", + "objective": "Every sanctioned path that touches the phone (android_ui, phone_deploy.sh) is mutually exclusive across Claude sessions, with automatic expiry so no session has to remember to release, and a refusal that names the holder and the alternatives.", + "acceptance_criteria": [ + "A second session's android_ui call against a phone leased by another session waits up to 60 s then exits 3 naming the holder", + "The same session's calls refresh the lease without waiting", + "A lease untouched for 180 s no longer blocks anyone", + "python_pkg.phone_lease and python_pkg.android_ui keep 100% line and branch coverage" + ], + "out_of_scope": [ + "Making two sessions drive the same screen concurrently (virtual display / emulator scripts cover that)", + "Raw adb commands, which bypass the tools" + ], + "verifier": "pre-commit run --files (pytest with --cov-fail-under=100), plus the two-session refusal reproduced on the real phone" +} diff --git a/docs/superpowers/evidence/cap-artifact-growth-2026-08-23.json b/docs/superpowers/evidence/cap-artifact-growth-2026-08-23.json deleted file mode 100644 index 6a2a9067..00000000 --- a/docs/superpowers/evidence/cap-artifact-growth-2026-08-23.json +++ /dev/null @@ -1,37 +0,0 @@ -{ - "intent": "Stop the superpowers bookkeeping tree from silently regrowing to the 695 files that made a 655-file prune necessary, by capping it in the gate instead of leaving a note recommending it.", - "scope": [ - "meta/scripts/check_repo_size.sh - adds a per-directory artifact cap", - "Non-goal: changing the source-file ceiling or the excluded prefixes, both unchanged." - ], - "changes": [ - "Added ARTIFACT_DIRS + MAX_ARTIFACTS_PER_DIR=60, checked by a new check_artifact_dirs helper wired into main alongside the existing source-file ceiling.", - "main now accumulates failures rather than returning on the first, so a commit that trips both the source ceiling and the artifact cap reports both.", - "60 is the headroom the prune bought (21 files per dir today). The header says to prune rather than raise it when it fails." - ], - "verification": [ - { - "command": "./meta/scripts/check_repo_size.sh", - "result": "pass", - "evidence": "'1050 source files (ceiling 1200)' EXIT=0 with evidence/contracts at 21 each, under the 60 cap. --count still prints the bare number, EXIT=0." - }, - { - "command": "MAX_ARTIFACTS_PER_DIR=5 variant", - "result": "pass", - "evidence": "EXIT=1 listing BOTH dirs (evidence 22, contracts 21) plus the prune hint - confirms the new branch fires and that it does not stop at the first offender." - }, - { - "command": "pre-commit run repo-size-cap --all-files; pre-commit run --files meta/scripts/check_repo_size.sh", - "result": "pass", - "evidence": "Hook Passed; shfmt and shellcheck clean on the modified script." - } - ], - "risks": [ - "A legitimate burst of commits can now trip the cap and block a commit until artifacts are pruned. That is the intended trade: the previous behaviour was unbounded growth noticed only months later.", - "The cap counts tracked files per directory, so it does not distinguish a needed recent artifact from an old one; the operator chooses what to prune." - ], - "rollback": [ - "git revert the commit; the source-file ceiling and all other behaviour are untouched by this change.", - "After rollback, confirm ./meta/scripts/check_repo_size.sh still exits 0 and prints the source count." - ] -} diff --git a/docs/superpowers/evidence/phone-lease-2026-09-20.json b/docs/superpowers/evidence/phone-lease-2026-09-20.json new file mode 100644 index 00000000..7008e3e8 --- /dev/null +++ b/docs/superpowers/evidence/phone-lease-2026-09-20.json @@ -0,0 +1,33 @@ +{ + "intent": "Two Claude sessions can no longer drive the phone at the same time: every android_ui call and every phone_deploy.sh run takes a per-serial lease owned by the session, and a foreign holder is refused by name with the alternatives spelled out.", + "scope": [ + "python_pkg/phone_lease (new): lease file, owner identity, acquire/release/status, CLI", + "python_pkg/android_ui/cli.py: acquire the lease before every command (exit 3 when held elsewhere)", + "Non-goal: parallel driving of one screen; that is what the virtual display / emulator scripts in ~/.claude/scripts are for" + ], + "changes": [ + "flock-guarded JSON lease per serial under ~/.cache/phone-lease/, 180 s TTL refreshed on every call", + "owner = nearest ancestor `claude` process (one per session); PHONE_LEASE_OWNER overrides", + "android_ui main() refuses with exit 3 and the holder's name; phone_deploy.sh exits 40" + ], + "verification": [ + { + "command": "pre-commit run --files python_pkg/phone_lease/... python_pkg/android_ui/...", + "result": "pass", + "evidence": "all hooks passed; python_pkg 1886 passed, 100% line+branch coverage" + }, + { + "command": "python3 -m python_pkg.android_ui -s 23181JEGR08034 focus; PHONE_LEASE_OWNER=other python3 -m python_pkg.android_ui -s 23181JEGR08034 focus", + "result": "pass", + "evidence": "first call refreshed claude:4102845's lease; the foreign call waited 60 s and exited 3: 'phone 23181JEGR08034 held by claude:4102845 (android_ui focus) until 17:32:00'" + } + ], + "risks": [ + "A session that keeps calling android_ui holds the phone indefinitely (by design); `python3 -m python_pkg.phone_lease release ` drops it", + "Raw adb use bypasses the lease; only the sanctioned tools take it" + ], + "rollback": [ + "Revert this commit; android_ui returns to lease-free operation", + "Validate: python3 -m python_pkg.android_ui focus runs without touching ~/.cache/phone-lease" + ] +} diff --git a/python_pkg/android_ui/README.md b/python_pkg/android_ui/README.md index 5f292e92..1c5176e9 100644 --- a/python_pkg/android_ui/README.md +++ b/python_pkg/android_ui/README.md @@ -87,3 +87,12 @@ ui.wait_for("Connected.", timeout=30) just gone empty. - Scrolling is not yet modelled: a widget outside the viewport is not in the tree. Scroll with `adb shell input swipe` first, or add `scroll_to()`. + +## One phone, many sessions + +Every command first takes this session's lease on the phone +(`python_pkg.phone_lease`): the owner is the nearest ancestor `claude` +process, the lease lasts 180 s past the last call, and a call from another +session waits up to 60 s, then exits 3 naming the holder. Do not retry in a +loop — use the phone's virtual display (`~/.claude/scripts/phone_vd.sh`) or +the headless emulator (`~/.claude/scripts/phone_emu.sh`), or ask the user. diff --git a/python_pkg/android_ui/cli.py b/python_pkg/android_ui/cli.py index 3872b345..b0fe5a95 100644 --- a/python_pkg/android_ui/cli.py +++ b/python_pkg/android_ui/cli.py @@ -17,6 +17,7 @@ import argparse import sys from python_pkg.android_ui.driver import AndroidUi, UiAutomationError +from python_pkg.phone_lease import PhoneBusyError, acquire def _build_parser() -> argparse.ArgumentParser: @@ -55,30 +56,51 @@ def _build_parser() -> argparse.ArgumentParser: return parser +def _lease_phone(serial: str | None, command: str) -> bool: + """Refresh this session's lease on the phone; False when another holds it. + + Every command goes through here first, so a second session's taps wait + instead of landing in whatever this one has in front (2026-09-20: a + sandbox tap opened another session's manga reader). Serial-less calls + share one key: adb picks one device anyway. + """ + try: + acquire(serial or "default", f"android_ui {command}") + except PhoneBusyError as exc: + sys.stderr.write(f"android-ui: {exc}\n") + return False + return True + + +def _run(ui: AndroidUi, args: argparse.Namespace) -> None: + """Dispatch one parsed command to the driver.""" + if args.command == "dump": + for element in ui.dump(): + sys.stdout.write(f"{element}\n") + elif args.command == "find": + sys.stdout.write(f"{ui.find(args.query, exact=args.exact)}\n") + elif args.command == "tap": + found = ui.tap(args.query, exact=args.exact, timeout=args.timeout) + sys.stdout.write(f"tapped {found}\n") + elif args.command == "wait": + found = ui.wait_for(args.query, timeout=args.timeout, exact=args.exact) + sys.stdout.write(f"{found}\n") + elif args.command == "type": + ui.type_into(args.query, args.text, exact=args.exact, timeout=args.timeout) + sys.stdout.write(f"typed into {args.query!r} and verified\n") + elif args.command == "dismiss-keyboard": + ui.dismiss_keyboard() + elif args.command == "focus": + sys.stdout.write(f"{ui.current_focus()}\n") + + def main(argv: list[str] | None = None) -> int: """Run the CLI. Returns a process exit code.""" args = _build_parser().parse_args(argv) - ui = AndroidUi(serial=args.serial) - + if not _lease_phone(args.serial, args.command): + return 3 try: - if args.command == "dump": - for element in ui.dump(): - sys.stdout.write(f"{element}\n") - elif args.command == "find": - sys.stdout.write(f"{ui.find(args.query, exact=args.exact)}\n") - elif args.command == "tap": - found = ui.tap(args.query, exact=args.exact, timeout=args.timeout) - sys.stdout.write(f"tapped {found}\n") - elif args.command == "wait": - found = ui.wait_for(args.query, timeout=args.timeout, exact=args.exact) - sys.stdout.write(f"{found}\n") - elif args.command == "type": - ui.type_into(args.query, args.text, exact=args.exact, timeout=args.timeout) - sys.stdout.write(f"typed into {args.query!r} and verified\n") - elif args.command == "dismiss-keyboard": - ui.dismiss_keyboard() - elif args.command == "focus": - sys.stdout.write(f"{ui.current_focus()}\n") + _run(AndroidUi(serial=args.serial), args) except UiAutomationError as exc: # Loud and non-zero on purpose: a silent no-op here is exactly the # failure this package exists to eliminate. The message already names diff --git a/python_pkg/android_ui/tests/test_cli.py b/python_pkg/android_ui/tests/test_cli.py index cad664b2..2261d98b 100644 --- a/python_pkg/android_ui/tests/test_cli.py +++ b/python_pkg/android_ui/tests/test_cli.py @@ -17,6 +17,7 @@ from python_pkg.android_ui.driver import ( ElementNotFoundError, UiElement, ) +from python_pkg.phone_lease import PhoneBusyError if TYPE_CHECKING: from collections.abc import Iterator @@ -39,13 +40,38 @@ def _element(label: str = "Connect Firebase") -> UiElement: @pytest.fixture def ui() -> Iterator[MagicMock]: - """Patch out the real device driver.""" - with patch(f"{MOD}.AndroidUi") as factory: + """Patch out the real device driver and the phone lease.""" + with patch(f"{MOD}.AndroidUi") as factory, patch(f"{MOD}.acquire"): instance = MagicMock() factory.return_value = instance yield instance +class TestLease: + def test_every_command_refreshes_the_lease_first(self) -> None: + with patch(f"{MOD}.AndroidUi"), patch(f"{MOD}.acquire") as acquire: + assert cli.main(["-s", "SER1", "focus"]) == 0 + acquire.assert_called_once_with("SER1", "android_ui focus") + + def test_serial_less_calls_share_one_key(self) -> None: + with patch(f"{MOD}.AndroidUi"), patch(f"{MOD}.acquire") as acquire: + cli.main(["focus"]) + assert acquire.call_args.args[0] == "default" + + def test_a_foreign_lease_stops_the_command( + self, capsys: pytest.CaptureFixture[str] + ) -> None: + with ( + patch(f"{MOD}.AndroidUi") as factory, + patch( + f"{MOD}.acquire", side_effect=PhoneBusyError("phone X held by claude:1") + ), + ): + assert cli.main(["-s", "X", "tap", "Save"]) == 3 + factory.assert_not_called() + assert "held by claude:1" in capsys.readouterr().err + + class TestDump: def test_prints_every_element( self, ui: MagicMock, capsys: pytest.CaptureFixture[str] diff --git a/python_pkg/phone_lease/__init__.py b/python_pkg/phone_lease/__init__.py new file mode 100644 index 00000000..b0c520ef --- /dev/null +++ b/python_pkg/phone_lease/__init__.py @@ -0,0 +1,41 @@ +"""One phone, many Claude sessions: a lease that keeps them from colliding. + +Two sessions driving the same phone interleave their taps -- on 2026-09-20 a +tap meant for the workout sandbox landed in another session's manga reader, +twice. This package hands the phone to one owner at a time: + +* the lease is a JSON file per serial under ``~/.cache/phone-lease/``, guarded + by ``flock`` so read-modify-write is atomic across processes; +* the owner is the nearest ancestor ``claude`` process (one per session), so + every call from the same session refreshes the same lease and calls from + another session wait, then fail naming the holder; +* it expires ``ttl`` seconds after the last refresh, so a session that just + stops driving frees the phone without anyone remembering to release. + +``android_ui`` takes the lease on every command and ``phone_deploy.sh`` on +every deploy, which is every path that touches the phone from a session. +""" + +from python_pkg.phone_lease.lease import ( + DEFAULT_TTL, + DEFAULT_WAIT, + Clock, + Lease, + PhoneBusyError, + acquire, + owner_id, + release, + status, +) + +__all__ = [ + "DEFAULT_TTL", + "DEFAULT_WAIT", + "Clock", + "Lease", + "PhoneBusyError", + "acquire", + "owner_id", + "release", + "status", +] diff --git a/python_pkg/phone_lease/__main__.py b/python_pkg/phone_lease/__main__.py new file mode 100644 index 00000000..d6f578d7 --- /dev/null +++ b/python_pkg/phone_lease/__main__.py @@ -0,0 +1,59 @@ +"""CLI: ``python3 -m python_pkg.phone_lease acquire|release|status ``.""" + +from __future__ import annotations + +import argparse +import sys + +from python_pkg.phone_lease.lease import ( + DEFAULT_TTL, + DEFAULT_WAIT, + PhoneBusyError, + acquire, + owner_id, + release, + status, +) + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="phone-lease", description=__doc__) + sub = parser.add_subparsers(dest="command", required=True) + take = sub.add_parser( + "acquire", help="take or refresh the lease (exit 3 when held elsewhere)" + ) + take.add_argument("serial") + take.add_argument( + "--note", default="", help="what the holder is doing, shown to the other side" + ) + take.add_argument("--ttl", type=float, default=DEFAULT_TTL) + take.add_argument("--wait", type=float, default=DEFAULT_WAIT) + drop = sub.add_parser("release", help="drop the lease if this session holds it") + drop.add_argument("serial") + look = sub.add_parser("status", help="print the holder, or 'free'") + look.add_argument("serial") + return parser + + +def main(argv: list[str] | None = None) -> int: + """Run the CLI. Exit 0 ok, 1 nothing to release, 3 phone held elsewhere.""" + args = _build_parser().parse_args(argv) + if args.command == "acquire": + try: + lease = acquire(args.serial, args.note, ttl=args.ttl, wait=args.wait) + except PhoneBusyError as exc: + sys.stderr.write(f"phone-lease: {exc}\n") + return 3 + sys.stdout.write(f"{lease.describe()}\n") + return 0 + if args.command == "release": + dropped = release(args.serial) + sys.stdout.write("released\n" if dropped else f"not held by {owner_id()}\n") + return 0 if dropped else 1 + lease = status(args.serial) + sys.stdout.write("free\n" if lease is None else f"{lease.describe()}\n") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/python_pkg/phone_lease/lease.py b/python_pkg/phone_lease/lease.py new file mode 100644 index 00000000..1e193e7a --- /dev/null +++ b/python_pkg/phone_lease/lease.py @@ -0,0 +1,181 @@ +"""The lease file, its owner identity, and the acquire/release operations.""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass +import fcntl +import json +import os +from pathlib import Path +import time +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Callable + + +@dataclass(frozen=True) +class Clock: + """The two things ``acquire`` needs from ``time``; tests hand in fakes.""" + + now: Callable[[], float] = time.time + sleep: Callable[[float], None] = time.sleep + + +# Seconds a lease outlives its last refresh; a quiet session frees the phone. +DEFAULT_TTL = 180.0 + +# Seconds ``acquire`` waits on a foreign lease before giving up. +DEFAULT_WAIT = 60.0 + +# Where lease files live; overridable so tests never touch the real one. +LEASE_DIR = Path( + os.environ.get("PHONE_LEASE_DIR", str(Path.home() / ".cache" / "phone-lease")) +) + +# Set this to name the owner explicitly (CI, a script without a claude parent). +OWNER_ENV = "PHONE_LEASE_OWNER" + + +class PhoneBusyError(RuntimeError): + """Raised when a fresh lease belongs to another owner.""" + + +@dataclass(frozen=True) +class Lease: + """What the lease file says.""" + + serial: str + owner: str + expires: float + note: str + + def is_expired(self, now: float) -> bool: + """True once ``now`` has passed the expiry.""" + return now >= self.expires + + def describe(self) -> str: + """One line for an error or a status readout.""" + until = time.strftime("%H:%M:%S", time.localtime(self.expires)) + return f"phone {self.serial} held by {self.owner} ({self.note}) until {until}" + + +def owner_id(pid: int | None = None, environ: dict[str, str] | None = None) -> str: + """Identify the calling session. + + ``PHONE_LEASE_OWNER`` wins. Otherwise walk up from ``pid`` (this process + by default) to the nearest ancestor whose ``comm`` is ``claude`` -- one + per session -- and fall back to this process's own pid when there is none. + """ + env = os.environ if environ is None else environ + explicit = env.get(OWNER_ENV) + if explicit: + return explicit + current = os.getpid() if pid is None else pid + origin = current + while current > 1: + try: + comm = Path(f"/proc/{current}/comm").read_text(encoding="utf-8").strip() + stat = Path(f"/proc/{current}/stat").read_text(encoding="utf-8") + except OSError: + break + if comm == "claude": + return f"claude:{current}" + # Field 4 of /proc//stat is the parent pid; the comm field before + # it may contain spaces, so split after its closing paren. + current = int(stat.rsplit(")", 1)[1].split()[1]) + return f"pid:{origin}" + + +def _paths(serial: str) -> tuple[Path, Path]: + LEASE_DIR.mkdir(parents=True, exist_ok=True) + safe = serial.replace("/", "_") + return LEASE_DIR / f"{safe}.lock", LEASE_DIR / f"{safe}.json" + + +def _read(path: Path) -> Lease | None: + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except OSError, ValueError: + return None + try: + return Lease(**raw) + except TypeError: + return None + + +def _locked_update( + serial: str, + owner: str, + note: str, + ttl: float, + now: float, +) -> Lease: + """One atomic attempt: take or refresh the lease, or return the holder's.""" + lock_path, lease_path = _paths(serial) + with lock_path.open("w", encoding="utf-8") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + current = _read(lease_path) + if ( + current is not None + and current.owner != owner + and not current.is_expired(now) + ): + return current + fresh = Lease(serial=serial, owner=owner, expires=now + ttl, note=note) + lease_path.write_text(json.dumps(asdict(fresh)), encoding="utf-8") + return fresh + + +def acquire( + serial: str, + note: str = "", + *, + owner: str | None = None, + ttl: float = DEFAULT_TTL, + wait: float = DEFAULT_WAIT, + clock: Clock | None = None, +) -> Lease: + """Take or refresh the lease on ``serial``. + + Waits up to ``wait`` seconds for a foreign lease to expire, polling once a + second, then raises :class:`PhoneBusyError` naming the holder. + """ + me = owner_id() if owner is None else owner + clock = clock or Clock() + deadline = clock.now() + wait + while True: + lease = _locked_update(serial, me, note, ttl, clock.now()) + if lease.owner == me: + return lease + if clock.now() >= deadline: + msg = ( + f"{lease.describe()} -- waited {wait:.0f}s. Do not retry in a loop: " + "drive a sandbox-flavor app on the phone's virtual display instead " + "(~/.claude/scripts/phone_vd.sh, needs no lease), or the headless " + "emulator (~/.claude/scripts/phone_emu.sh); see the phone-deploy skill." + ) + raise PhoneBusyError(msg) + clock.sleep(1.0) + + +def release(serial: str, *, owner: str | None = None) -> bool: + """Drop the lease if ``owner`` holds it. Returns whether anything was dropped.""" + me = owner_id() if owner is None else owner + lock_path, lease_path = _paths(serial) + with lock_path.open("w", encoding="utf-8") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + current = _read(lease_path) + if current is None or current.owner != me: + return False + lease_path.unlink() + return True + + +def status(serial: str, now: float | None = None) -> Lease | None: + """The live lease on ``serial``, or None when free or expired.""" + _, lease_path = _paths(serial) + current = _read(lease_path) + if current is None or current.is_expired(time.time() if now is None else now): + return None + return current diff --git a/python_pkg/phone_lease/tests/__init__.py b/python_pkg/phone_lease/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/python_pkg/phone_lease/tests/test_lease.py b/python_pkg/phone_lease/tests/test_lease.py new file mode 100644 index 00000000..eb510ebe --- /dev/null +++ b/python_pkg/phone_lease/tests/test_lease.py @@ -0,0 +1,193 @@ +"""The lease: owner identity, atomic take/refresh, expiry, release, status.""" + +from __future__ import annotations + +from pathlib import Path +import time +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from python_pkg.phone_lease import ( + Clock, + Lease, + PhoneBusyError, + acquire, + owner_id, + release, + status, +) +from python_pkg.phone_lease import lease as lease_mod + +if TYPE_CHECKING: + from collections.abc import Callable, Iterator + + +class FakeClock: + """A clock `acquire` can be handed instead of `time`.""" + + def __init__(self, now: float = 1000.0) -> None: + self.now = now + self.slept: list[float] = [] + + def time(self) -> float: + return self.now + + def sleep(self, secs: float) -> None: + self.slept.append(secs) + self.now += secs + + def as_clock(self) -> Clock: + return Clock(now=self.time, sleep=self.sleep) + + +@pytest.fixture(autouse=True) +def lease_dir(tmp_path: Path) -> Iterator[Path]: + """Every test gets its own lease directory.""" + with patch.object(lease_mod, "LEASE_DIR", tmp_path / "leases"): + yield tmp_path / "leases" + + +class TestOwnerId: + def test_explicit_env_wins(self) -> None: + assert owner_id(environ={"PHONE_LEASE_OWNER": "ci-run-7"}) == "ci-run-7" + + def test_finds_the_claude_ancestor(self, tmp_path: Path) -> None: + proc = tmp_path / "proc" + for pid, comm, ppid in ( + (300, "zsh", 200), + (200, "claude", 100), + (100, "systemd", 1), + ): + d = proc / str(pid) + d.mkdir(parents=True) + (d / "comm").write_text(f"{comm}\n") + (d / "stat").write_text(f"{pid} ({comm}) S {ppid} 1 1 0 -1\n") + with patch.object(lease_mod.Path, "read_text", _proc_reader(proc)): + assert owner_id(pid=300, environ={}) == "claude:200" + + def test_falls_back_to_own_pid_without_a_claude_ancestor( + self, tmp_path: Path + ) -> None: + proc = tmp_path / "proc" + d = proc / "300" + d.mkdir(parents=True) + (d / "comm").write_text("zsh\n") + (d / "stat").write_text("300 (zsh) S 1 1 1 0 -1\n") + with patch.object(lease_mod.Path, "read_text", _proc_reader(proc)): + assert owner_id(pid=300, environ={}) == "pid:300" + + def test_unreadable_proc_falls_back_to_own_pid(self) -> None: + assert owner_id(pid=2**22 + 12345, environ={}) == f"pid:{2**22 + 12345}" + + def test_defaults_to_this_process(self) -> None: + assert owner_id(environ={}).split(":")[0] in {"claude", "pid"} + + +def _proc_reader(proc: Path) -> Callable[..., str]: + real = Path.read_text + + def read_text(self: Path, *args: object, **kwargs: object) -> str: + if str(self).startswith("/proc/"): + return real(proc / str(self)[len("/proc/") :], *args, **kwargs) + return real(self, *args, **kwargs) + + return read_text + + +class TestAcquire: + def test_takes_a_free_phone(self, lease_dir: Path) -> None: + clock = FakeClock() + got = acquire( + "PIXEL", "deploy", owner="claude:1", ttl=10, clock=clock.as_clock() + ) + assert got == Lease( + serial="PIXEL", owner="claude:1", expires=1010.0, note="deploy" + ) + assert (lease_dir / "PIXEL.json").exists() + + def test_refreshes_the_same_owner(self) -> None: + clock = FakeClock() + acquire("PIXEL", "a", owner="claude:1", ttl=10, clock=clock.as_clock()) + clock.now += 5 + got = acquire("PIXEL", "b", owner="claude:1", ttl=10, clock=clock.as_clock()) + assert got.expires == 1015.0 + assert got.note == "b" + assert clock.slept == [] + + def test_waits_for_a_foreign_lease_to_expire(self) -> None: + clock = FakeClock() + acquire("PIXEL", "other", owner="claude:2", ttl=3, clock=clock.as_clock()) + got = acquire( + "PIXEL", "mine", owner="claude:1", ttl=10, wait=60, clock=clock.as_clock() + ) + assert got.owner == "claude:1" + assert clock.slept == [1.0, 1.0, 1.0] + + def test_gives_up_naming_the_holder(self) -> None: + clock = FakeClock() + acquire("PIXEL", "manga", owner="claude:2", ttl=1000, clock=clock.as_clock()) + with pytest.raises( + PhoneBusyError, match=r"held by claude:2 \(manga\).*waited 2s" + ): + acquire("PIXEL", "mine", owner="claude:1", wait=2, clock=clock.as_clock()) + + def test_uses_the_real_clock_by_default(self) -> None: + got = acquire("PIXEL", owner="claude:1", ttl=5) + assert got.expires == pytest.approx(time.time() + 5, abs=2) + + def test_uses_the_real_owner_by_default(self) -> None: + assert acquire("PIXEL").owner == owner_id() + + def test_serial_with_slash_is_a_safe_filename(self, lease_dir: Path) -> None: + acquire("192.168.1.5:5555/x", owner="claude:1") + assert (lease_dir / "192.168.1.5:5555_x.json").exists() + + +class TestCorruptLeaseFile: + def test_garbage_is_treated_as_free(self, lease_dir: Path) -> None: + lease_dir.mkdir() + (lease_dir / "PIXEL.json").write_text("not json") + assert acquire("PIXEL", owner="claude:1").owner == "claude:1" + + def test_wrong_shape_is_treated_as_free(self, lease_dir: Path) -> None: + lease_dir.mkdir() + (lease_dir / "PIXEL.json").write_text('{"unexpected": 1}') + assert status("PIXEL") is None + + +class TestRelease: + def test_drops_own_lease(self) -> None: + acquire("PIXEL", owner="claude:1") + assert release("PIXEL", owner="claude:1") is True + assert status("PIXEL") is None + + def test_leaves_a_foreign_lease_alone(self) -> None: + acquire("PIXEL", owner="claude:2") + assert release("PIXEL", owner="claude:1") is False + assert status("PIXEL") is not None + + def test_nothing_to_release(self) -> None: + assert release("PIXEL", owner="claude:1") is False + + def test_defaults_to_this_owner(self) -> None: + acquire("PIXEL") + assert release("PIXEL") is True + + +class TestStatus: + def test_free(self) -> None: + assert status("PIXEL") is None + + def test_expired_is_free(self) -> None: + acquire("PIXEL", owner="claude:1", ttl=1, clock=FakeClock(now=100.0).as_clock()) + assert status("PIXEL", now=200.0) is None + + def test_live_describes_the_holder(self) -> None: + acquire("PIXEL", "deploy", owner="claude:1", ttl=1000) + live = status("PIXEL") + assert live is not None + assert live.describe().startswith( + "phone PIXEL held by claude:1 (deploy) until " + ) diff --git a/python_pkg/phone_lease/tests/test_main.py b/python_pkg/phone_lease/tests/test_main.py new file mode 100644 index 00000000..e6ac0eb8 --- /dev/null +++ b/python_pkg/phone_lease/tests/test_main.py @@ -0,0 +1,57 @@ +"""The CLI wrapper: exit codes and the one-line outputs.""" + +from __future__ import annotations + +from pathlib import Path +import runpy +import sys +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from python_pkg.phone_lease import __main__ as cli +from python_pkg.phone_lease import lease as lease_mod + +if TYPE_CHECKING: + from collections.abc import Iterator + + +@pytest.fixture(autouse=True) +def lease_dir(tmp_path: Path) -> Iterator[None]: + with patch.object(lease_mod, "LEASE_DIR", tmp_path / "leases"): + yield + + +def test_acquire_then_status_then_release(capsys: pytest.CaptureFixture[str]) -> None: + assert cli.main(["acquire", "PIXEL", "--note", "deploy"]) == 0 + assert "held by" in capsys.readouterr().out + assert cli.main(["status", "PIXEL"]) == 0 + assert "(deploy)" in capsys.readouterr().out + assert cli.main(["release", "PIXEL"]) == 0 + assert capsys.readouterr().out == "released\n" + assert cli.main(["status", "PIXEL"]) == 0 + assert capsys.readouterr().out == "free\n" + + +def test_release_without_holding(capsys: pytest.CaptureFixture[str]) -> None: + assert cli.main(["release", "PIXEL"]) == 1 + assert "not held by" in capsys.readouterr().out + + +def test_foreign_holder_exits_3(capsys: pytest.CaptureFixture[str]) -> None: + lease_mod.acquire("PIXEL", "manga", owner="someone-else", ttl=1000) + assert cli.main(["acquire", "PIXEL", "--wait", "0"]) == 3 + assert "held by someone-else (manga)" in capsys.readouterr().err + + +def test_module_entry_point() -> None: + # runpy warns if __main__ is already imported (this file imports it), so + # drop it first; the test then runs the real module top to bottom. + sys.modules.pop("python_pkg.phone_lease.__main__", None) + with ( + patch("sys.argv", ["phone-lease", "status", "PIXEL"]), + pytest.raises(SystemExit) as excinfo, + ): + runpy.run_module("python_pkg.phone_lease", run_name="__main__") + assert excinfo.value.code == 0