TUI sub-phase θ — binding interface SPEC

1async def _run_streamed(cmd: list[str]) -> int:2    """Async streaming variant of the previous blocking _run_sync. stdout is3    line-buffered to OUR stdout so the TUI tails it live; rc is returned."""4    proc = await asyncio.create_subprocess_exec(5        *cmd,6        stdout=asyncio.subprocess.PIPE,7        stderr=asyncio.subprocess.STDOUT,8    )9    assert proc.stdout is not None10    while True:11        line = await proc.stdout.readline()12        if not line:13            break14        sys.stdout.write(line.decode(errors="replace"))15        sys.stdout.flush()16    return await proc.wait()

1async def _drive(args):2    try:3        parse_pr_url(args.pr_url)4    except Exception as exc:5        _emit(f"(error: invalid pr_url: {exc})")6        return 278    adk_bin = ...; agent_bin = ...; queue = ...910    agent_proc: asyncio.subprocess.Process | None = None11    sig_received = False1213    def _on_sigterm() -> None:14        nonlocal sig_received15        sig_received = True16        if agent_proc is not None and agent_proc.returncode is None:17            try:18                agent_proc.terminate()19            except ProcessLookupError:20                pass2122    try:23        asyncio.get_running_loop().add_signal_handler(signal.SIGTERM, _on_sigterm)24    except NotImplementedError:  # pragma: no cover25        pass2627    # Now claim/prepare/release calls run with the handler in place.28    ...

1rc = await _run_streamed([adk_bin, "pr-queue", "claim", args.pr_url, "--queue", queue])2if rc != 0:3    _emit("(error: claim failed — row may be locked by another reviewer)")4    return 25_emit(f"(claimed: {args.pr_url})")6if sig_received:7    await _release(adk_bin, args.pr_url, queue, status="error")8    return 130910if not args.no_prepare:11    prep_cmd = [adk_bin, "pr-task", "prepare", args.pr_url, "--queue", queue]12    _emit("$ " + " ".join(prep_cmd))13    rc = await _run_streamed(prep_cmd)14    if rc != 0:15        _emit(f"(error: prepare failed rc={rc})")16        await _release(adk_bin, args.pr_url, queue, status="error")17        return 118    if sig_received:19        await _release(adk_bin, args.pr_url, queue, status="error")20        return 130

1async def _release(adk_bin: str, pr_url: str, queue: str, *, status: str | None = None) -> None:2    cmd = [adk_bin, "pr-queue", "release", pr_url, "--queue", queue]3    if status:4        cmd += ["--status", status]5    rc = await _run_streamed(cmd)6    if rc == 0:7        _emit(f"(released: {pr_url})")8    else:9        _emit(f"(error: release rc={rc} — lock may be stuck)")

1python3 tui/worker.py https://github.com/foo/bar/pull/200 \2  --adk-bin /tmp/fake-adk --agent-bin /tmp/fake-claude \3  --queue /tmp/q --heartbeat-dir /tmp/w \4  --no-prepare --heartbeat-bump-interval-s 0.3 --heartbeat-file-interval-s 0.1

1from __future__ import annotations23import json4from dataclasses import dataclass5from datetime import datetime, timezone6from pathlib import Path7from typing import Callable8910@dataclass(frozen=True)11class WorkerRow:12    pid: int13    pr_url: str14    task_type: str15    agent: str16    queue: str17    started_at: str18    last_heartbeat: str19    current_phase: str20    rc: int | None21    age_s: float22    is_stale: bool232425def default_workers_dir() -> Path:26    return Path.home() / ".agents-devkit" / "tui" / "workers"272829def _parse_iso(ts: str | None) -> datetime | None:30    if not ts:31        return None32    try:33        if ts.endswith("Z"):34            ts = ts[:-1] + "+00:00"35        return datetime.fromisoformat(ts)36    except ValueError:37        return None383940class WorkersModel:41    """Scans a directory of <pid>.json heartbeat files. Returns the live ones,42    GCs the dead. Mtime-gated by directory mtime AND by aggregate file mtime43    (to catch a single file's update without a directory mutation)."""4445    def __init__(46        self,47        workers_dir: Path | None = None,48        *,49        stale_after_s: float = 30.0,50        gc_after_s: float = 120.0,51        now_fn: Callable[[], datetime] | None = None,52    ) -> None:53        self.workers_dir = workers_dir if workers_dir is not None else default_workers_dir()54        self.stale_after_s = stale_after_s55        self.gc_after_s = gc_after_s56        if now_fn is None:57            now_fn = lambda: datetime.now(tz=timezone.utc)  # noqa: E73158        self._now_fn = now_fn59        self._last_signature: tuple | None = None6061    def _signature(self) -> tuple:62        """Cheap directory fingerprint. (dir_mtime, [(name, mtime, size)...])."""63        if not self.workers_dir.exists():64            return (0.0, ())65        try:66            dm = self.workers_dir.stat().st_mtime67        except OSError:68            return (0.0, ())69        items: list[tuple[str, float, int]] = []70        try:71            for p in self.workers_dir.iterdir():72                if p.suffix == ".json":73                    try:74                        st = p.stat()75                        items.append((p.name, st.st_mtime, st.st_size))76                    except OSError:77                        continue78        except OSError:79            return (dm, ())80        items.sort()81        return (dm, tuple(items))8283    def has_changed(self) -> bool:84        cur = self._signature()85        if cur != self._last_signature:86            return True87        return False8889    def snapshot(self) -> list[WorkerRow]:90        self._last_signature = self._signature()91        if not self.workers_dir.exists():92            return []93        now = self._now_fn()94        rows: list[WorkerRow] = []95        for p in sorted(self.workers_dir.iterdir()):96            if p.suffix != ".json":97                continue98            row = self._parse_one(p, now)99            if row is None:100                continue101            rows.append(row)102        return rows103104    def gc(self) -> int:105        """Remove heartbeat files older than gc_after_s. Returns the count removed."""106        if not self.workers_dir.exists():107            return 0108        now = self._now_fn()109        removed = 0110        for p in self.workers_dir.iterdir():111            if p.suffix != ".json":112                continue113            try:114                raw = json.loads(p.read_text())115                last = _parse_iso(raw.get("last_heartbeat"))116            except (OSError, json.JSONDecodeError, TypeError):117                last = None118            if last is None:119                # Unparseable file — GC if its mtime is also old enough.120                try:121                    file_age = (now.timestamp() - p.stat().st_mtime)122                except OSError:123                    continue124                if file_age > self.gc_after_s:125                    self._unlink(p)126                    removed += 1127                continue128            age = (now - last).total_seconds()129            if age > self.gc_after_s:130                self._unlink(p)131                removed += 1132        return removed133134    def _parse_one(self, p: Path, now: datetime) -> WorkerRow | None:135        try:136            raw = json.loads(p.read_text())137        except (OSError, json.JSONDecodeError):138            return None139        last = _parse_iso(raw.get("last_heartbeat"))140        if last is None:141            return None142        age = (now - last).total_seconds()143        is_stale = age > self.stale_after_s144        try:145            return WorkerRow(146                pid=int(raw.get("pid", 0)),147                pr_url=str(raw.get("pr_url", "")),148                task_type=str(raw.get("task_type", "")),149                agent=str(raw.get("agent", "")),150                queue=str(raw.get("queue", "")),151                started_at=str(raw.get("started_at", "")),152                last_heartbeat=str(raw.get("last_heartbeat", "")),153                current_phase=str(raw.get("current_phase", "")),154                rc=raw.get("rc"),155                age_s=age,156                is_stale=is_stale,157            )158        except (TypeError, ValueError):159            return None160161    @staticmethod162    def _unlink(p: Path) -> None:163        try:164            p.unlink()165        except OSError:166            pass

1from __future__ import annotations23from textual.widgets import Static45from tui.model.workers_model import WorkerRow678class WorkersPane(Static):9    def __init__(self) -> None:10        super().__init__("(no active workers)", markup=False)1112    def update_workers(self, rows: list[WorkerRow], *, ascii_only: bool = False) -> None:13        # Stale rows are hidden by default (the user only sees live workers).14        live = [r for r in rows if not r.is_stale]15        if not live:16            self.update("(no active workers)")17            return18        header = f"Workers ({len(live)} active)"19        body = [_format_row(r, ascii_only=ascii_only) for r in live]20        self.update("\n".join([header, *body]))212223def _format_row(row: WorkerRow, *, ascii_only: bool) -> str:24    pr_short = _shorten(row.pr_url)25    age = _format_age(row.age_s)26    glyph = "⚙↻" if not ascii_only else "~"27    return f"  {glyph}  {pr_short}  ·  {row.task_type}/{row.current_phase}  ·  {row.agent}  ·  {age}"282930def _shorten(pr_url: str) -> str:31    # https://github.com/acme/foo/pull/42 → acme/foo#4232    # https://bitbucket.org/o/r/pull-requests/5 → o/r#533    import re34    m = re.search(r"github\.com/([^/]+)/([^/]+)/pull/(\d+)", pr_url)35    if m:36        return f"{m.group(1)}/{m.group(2)}#{m.group(3)}"37    m = re.search(r"bitbucket\.org/([^/]+)/([^/]+)/pull-requests/(\d+)", pr_url)38    if m:39        return f"{m.group(1)}/{m.group(2)}#{m.group(3)}"40    return pr_url414243def _format_age(seconds: float) -> str:44    if seconds < 60:45        return f"{int(seconds)}s"46    minutes = int(seconds // 60)47    if minutes < 60:48        return f"{minutes}m"49    hours = int(minutes // 60)50    if hours < 24:51        return f"{hours}h"52    return f"{int(hours // 24)}d"

1WorkersPane {2    height: auto;3    min-height: 1;4    max-height: 10;5    padding: 0 1;6    background: $boost;7}

1from tui.widgets.workers_pane import WorkersPane2if TYPE_CHECKING:3    from tui.model.workers_model import WorkersModel

1self._workers_model: WorkersModel | None = None2self._workers_dir_override: Path | None = None  # populated from heartbeat_dir kwarg

1def compose(self) -> ComposeResult:2    yield HeaderBar()3    with Horizontal(id="main"):4        yield QueueTable()5        yield DetailPane()6    yield WorkersPane()7    yield SyncPlanPane()8    yield LogPane()9    yield FooterBar()

1async def on_mount(self) -> None:2    from tui.model.queue_model import QueueModel3    from tui.model.sync_plan_model import SyncPlanModel4    from tui.model.workers_model import WorkersModel56    self._model = QueueModel(queue_path=self._queue_path)7    self._plan_model = SyncPlanModel(plan_path=self._plan_path)8    self._workers_model = WorkersModel(workers_dir=self._heartbeat_dir)9    self.query_one(FooterBar).update_status(self._filter_mode, self._sort_mode)10    self._reload(force=True)11    self._reload_plan(force=True)12    self._reload_workers(force=True)13    self.set_interval(self.poll_interval, self._maybe_reload)

1def _reload_workers(self, *, force: bool = False) -> None:2    if self._workers_model is None:3        return4    if not force and not self._workers_model.has_changed():5        return6    rows = self._workers_model.snapshot()7    self.query_one(WorkersPane).update_workers(rows, ascii_only=self._ascii_only)

1def _maybe_reload(self) -> None:2    if self._model is not None and self._model.has_changed():3        self._reload(force=True)4    self._reload_plan()5    self._reload_workers()

1@pytest.fixture2def workers_dir_with_two(tmp_path: Path) -> Path:3    """A workers/ dir containing two fresh heartbeat files."""4    import json5    d = tmp_path / "workers"6    d.mkdir(parents=True)7    now_iso = "2026-05-22T14:00:00Z"8    for pid, pr in [(11111, "https://github.com/acme/foo/pull/42"),9                    (22222, "https://github.com/acme/bar/pull/7")]:10        (d / f"{pid}.json").write_text(json.dumps({11            "pid": pid, "pr_url": pr, "task_type": "review",12            "agent": "claude", "queue": "/tmp/q",13            "started_at": now_iso, "last_heartbeat": now_iso,14            "current_phase": "review", "rc": None,15        }))16    return d171819@pytest.fixture20def stale_worker_file(tmp_path: Path) -> Path:21    """A workers/ dir with a single STALE heartbeat (5 min ago)."""22    import json23    d = tmp_path / "workers"24    d.mkdir(parents=True)25    old_iso = "2026-05-22T13:55:00Z"26    (d / "99999.json").write_text(json.dumps({27        "pid": 99999, "pr_url": "https://github.com/acme/old/pull/1",28        "task_type": "review", "agent": "claude", "queue": "/tmp/q",29        "started_at": old_iso, "last_heartbeat": old_iso,30        "current_phase": "review", "rc": None,31    }))32    return d333435@pytest.fixture36def fake_slow_adk_script(tmp_path: Path) -> Path:37    """A /bin/sh adk that takes >1s to complete each verb — used by the SIGTERM38    streaming tests. Reads $ADK_SLOW_S (default 2) for sleep duration."""39    p = tmp_path / "slow-adk"40    p.write_text(41        "#!/bin/sh\n"42        "echo \"slow-adk $@\"\n"43        "sleep \"${ADK_SLOW_S:-2}\"\n"44        "echo ok\n"45        "exit 0\n"46    )47    p.chmod(0o755)48    return p

1deadline = time.time() + 3.02while time.time() < deadline:3    if list(worker_heartbeat_dir.glob("*.json")):4        break5    time.sleep(0.05)6proc.send_signal(signal.SIGTERM)

1deadline = time.time() + 3.02while time.time() < deadline:3    if list(worker_heartbeat_dir.glob("*.json")):4        break5    time.sleep(0.05)6else:7    proc.terminate()  # belt and suspenders cleanup8    proc.wait(timeout=2)9    pytest.fail("heartbeat file never appeared within 3 s")10proc.send_signal(signal.SIGTERM)