TUI sub-phase γ — binding interface SPEC

1{2  "version": 1,3  "queue": "/Users/sujeet/.agents-devkit/config/pr-queue.json5",4  "argv": ["--no-scan"],5  "started_at": "2026-05-22T14:32:15Z",6  "updated_at": "2026-05-22T14:33:01Z",7  "completed_at": null,8  "rc": null,9  "steps": [10    {11      "name": "pr-scan",12      "status": "ok",13      "rc": 0,14      "started_at": "2026-05-22T14:32:15Z",15      "completed_at": "2026-05-22T14:32:42Z"16    },17    {18      "name": "pr-queue update --all",19      "status": "running",20      "rc": null,21      "started_at": "2026-05-22T14:32:42Z",22      "completed_at": null23    },24    {25      "name": "pr-queue clean (merged)",26      "status": "pending",27      "rc": null,28      "started_at": null,29      "completed_at": null30    }31  ]32}

1"""TUI sync-plan writer. Persists ~/.agents-devkit/tui/workers/sync-plan.json2during pr-sync execution so the TUI's Sync-plan pane can render live progress.3"""4from __future__ import annotations56import json7import os8from dataclasses import dataclass, field, asdict9from pathlib import Path10from typing import Literal1112PLAN_VERSION = 11314DEFAULT_PLAN_PATH = Path.home() / ".agents-devkit" / "tui" / "workers" / "sync-plan.json"1516StepStatus = Literal["pending", "running", "ok", "warn", "failed", "skipped"]171819def plan_path() -> Path:20    """Return ADK_TUI_PLAN_PATH if set, else DEFAULT_PLAN_PATH."""21    env = os.environ.get("ADK_TUI_PLAN_PATH")22    return Path(env) if env else DEFAULT_PLAN_PATH232425@dataclass26class StepRecord:27    name: str28    status: StepStatus = "pending"29    rc: int | None = None30    started_at: str | None = None31    completed_at: str | None = None323334class SyncPlanWriter:35    """Append-only step tracker that persists to PLAN_PATH atomically."""3637    def __init__(self, queue: str, argv: list[str], step_names: list[str], *, path: Path | None = None) -> None:38        self.path = path if path is not None else plan_path()39        self._queue = queue40        self._argv = list(argv)41        self._steps: list[StepRecord] = [StepRecord(name=n) for n in step_names]42        self._started_at = _utc_now_iso()43        self._completed_at: str | None = None44        self._rc: int | None = None45        self._index_by_name: dict[str, int] = {n: i for i, n in enumerate(step_names)}46        self._write()  # initial snapshot — TUI sees plan as soon as pr-sync starts4748    def step_start(self, name: str) -> None:49        rec = self._get(name)50        rec.status = "running"51        rec.started_at = _utc_now_iso()52        self._write()5354    def step_done(self, name: str, *, status: StepStatus, rc: int | None = None) -> None:55        rec = self._get(name)56        rec.status = status57        rec.rc = rc58        rec.completed_at = _utc_now_iso()59        self._write()6061    def finish(self, rc: int) -> None:62        self._completed_at = _utc_now_iso()63        self._rc = rc64        self._write()6566    # --- internals ---67    def _get(self, name: str) -> StepRecord:68        idx = self._index_by_name.get(name)69        if idx is None:70            # Tolerant: append a new step on the fly. Keeps pr_sync evolution painless.71            rec = StepRecord(name=name)72            self._steps.append(rec)73            self._index_by_name[name] = len(self._steps) - 174            return rec75        return self._steps[idx]7677    def _write(self) -> None:78        payload = {79            "version": PLAN_VERSION,80            "queue": self._queue,81            "argv": self._argv,82            "started_at": self._started_at,83            "updated_at": _utc_now_iso(),84            "completed_at": self._completed_at,85            "rc": self._rc,86            "steps": [asdict(s) for s in self._steps],87        }88        self.path.parent.mkdir(parents=True, exist_ok=True)89        tmp = self.path.with_suffix(self.path.suffix + ".tmp")90        tmp.write_text(json.dumps(payload, indent=2))91        os.replace(tmp, self.path)929394def _utc_now_iso() -> str:95    import datetime as _dt96    return _dt.datetime.now(tz=_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

1pr-scan2pr-queue update --all3pr-queue clean (merged)4pr-task clean-orphans                       (or '… (dry-run)' suffix when --dry-run)5pr-queue remind                             (or '… (dry-run)' suffix when --dry-run)6base-index audit7auto-base cleanup8pr-task prepare --all

1from tui_plan import SyncPlanWriter  # NEW import (lazy in main, not module-level)23_PR_SYNC_STEPS = [4    "pr-scan",5    "pr-queue update --all",6    "pr-queue clean (merged)",7    "pr-task clean-orphans",8    "pr-queue remind",9    "base-index audit",10    "auto-base cleanup",11    "pr-task prepare --all",12]13plan_writer = SyncPlanWriter(queue=queue, argv=list(argv or []), step_names=_PR_SYNC_STEPS)

1def _run_step(name: str, fn, log, *, plan: SyncPlanWriter | None = None) -> dict:2    log.info("=== step: %s ===", name)3    if plan is not None:4        plan.step_start(name)5    try:6        rc = fn()7        status = "ok" if rc == 0 else "warn"8        if plan is not None:9            plan.step_done(name, status=status, rc=rc)10        return {"step": name, "rc": rc, "status": status}11    except SystemExit as e:12        log.warning("%s: %s", name, e)13        if plan is not None:14            plan.step_done(name, status="failed", rc=1)15        return {"step": name, "rc": 1, "status": "failed", "reason": str(e)}16    except Exception as e:17        log.warning("%s: unexpected %s", name, e)18        if plan is not None:19            plan.step_done(name, status="failed", rc=1)20        return {"step": name, "rc": 1, "status": "failed", "reason": str(e)}

1if args.no_scan:2    plan_writer.step_done("pr-scan", status="skipped")3    results.append({"step": "pr-scan", "status": "skipped"})4else:5    # … existing code, including _run_step(..., plan=plan_writer)

1rc_out = 1 if summary["failed"] else 02plan_writer.finish(rc_out)3return rc_out

1from __future__ import annotations23from textual.widgets import RichLog456class LogPane(RichLog):7    """Streaming output pane for `adk pr-sync` stdout."""89    def __init__(self) -> None:10        super().__init__(highlight=False, markup=False, wrap=False, auto_scroll=True)1112    def announce(self, msg: str) -> None:13        """Write a meta-line (start/exit/error). The pane doesn't distinguish14        types; the message itself includes its prefix."""15        self.write(msg)

1from __future__ import annotations23from textual.widgets import Static45from tui.model.sync_plan_model import SyncPlanSnapshot67_ICONS = {8    "pending":  ("…",  "[--]"),9    "running":  ("⚡", "[..]"),10    "ok":       ("✓",  "[ok]"),11    "warn":     ("⚠",  "[wn]"),12    "failed":   ("✗",  "[fl]"),13    "skipped":  ("↷",  "[sk]"),14}151617class SyncPlanPane(Static):18    def __init__(self) -> None:19        super().__init__("(no sync run yet — press `s` to start)", markup=False)2021    def update_snapshot(self, snapshot: SyncPlanSnapshot | None, *, ascii_only: bool = False) -> None:22        if snapshot is None:23            self.update("(no sync run yet — press `s` to start)")24            return25        # First line: header. Second line onwards: one step per row.26        header = _format_header(snapshot)27        body = [_format_step(s, ascii_only=ascii_only) for s in snapshot.steps]28        self.update("\n".join([header, *body]))293031def _format_header(snap: SyncPlanSnapshot) -> str:32    done = sum(1 for s in snap.steps if s.status in ("ok", "warn", "failed", "skipped"))33    total = len(snap.steps)34    if snap.completed_at is not None:35        outcome = "✓ done" if (snap.rc or 0) == 0 else "✗ done (with failures)"36        return f"Sync plan ({outcome} · {done}/{total} steps)"37    return f"Sync plan (running · {done}/{total} steps)"383940def _format_step(step, *, ascii_only: bool) -> str:41    icon = _ICONS[step.status][1 if ascii_only else 0]42    return f"  {icon}  {step.name}"

1from __future__ import annotations23import json4import os5from dataclasses import dataclass6from pathlib import Path789@dataclass(frozen=True)10class SyncPlanStep:11    name: str12    status: str13    rc: int | None14    started_at: str | None15    completed_at: str | None161718@dataclass(frozen=True)19class SyncPlanSnapshot:20    queue: str21    started_at: str22    updated_at: str23    completed_at: str | None24    rc: int | None25    steps: list[SyncPlanStep]262728def default_plan_path() -> Path:29    env = os.environ.get("ADK_TUI_PLAN_PATH")30    if env:31        return Path(env)32    return Path.home() / ".agents-devkit" / "tui" / "workers" / "sync-plan.json"333435class SyncPlanModel:36    def __init__(self, plan_path: Path | None = None) -> None:37        self.plan_path = plan_path if plan_path is not None else default_plan_path()38        self._last_mtime: float | None = None3940    def has_changed(self) -> bool:41        if not self.plan_path.exists():42            return self._last_mtime not in (None, 0.0) or self._last_mtime is None43        try:44            cur = self.plan_path.stat().st_mtime45        except OSError:46            return False47        return cur != self._last_mtime4849    def snapshot(self) -> SyncPlanSnapshot | None:50        if not self.plan_path.exists():51            self._last_mtime = 0.052            return None53        try:54            self._last_mtime = self.plan_path.stat().st_mtime55            raw = json.loads(self.plan_path.read_text())56        except (OSError, json.JSONDecodeError):57            return None58        if raw.get("version") != 1:59            return None60        try:61            steps = [62                SyncPlanStep(63                    name=str(s.get("name", "")),64                    status=str(s.get("status", "pending")),65                    rc=s.get("rc"),66                    started_at=s.get("started_at"),67                    completed_at=s.get("completed_at"),68                )69                for s in raw.get("steps") or []70            ]71            return SyncPlanSnapshot(72                queue=str(raw.get("queue", "")),73                started_at=str(raw.get("started_at", "")),74                updated_at=str(raw.get("updated_at", "")),75                completed_at=raw.get("completed_at"),76                rc=raw.get("rc"),77                steps=steps,78            )79        except (TypeError, ValueError):80            return None

1Screen {2    layout: vertical;3}45HeaderBar {6    dock: top;7    height: 1;8    background: $primary-background;9    color: $text;10    padding: 0 1;11}1213FooterBar {14    dock: bottom;15    height: 1;16    background: $primary-background;17    color: $text;18    padding: 0 1;19}2021#main {22    layout: horizontal;23    height: 1fr;24}2526QueueTable {27    width: 2fr;28    height: 1fr;29}3031QueueTable > .datatable--header {32    text-style: bold;33}3435DetailPane {36    width: 1fr;37    height: 1fr;38    padding: 1 2;39    background: $surface;40}4142SyncPlanPane {43    height: auto;44    min-height: 3;45    max-height: 12;46    padding: 0 1;47    background: $boost;48}4950LogPane {51    height: 14;52    border-top: solid $primary;53    background: $surface;54}

1import asyncio2import os3import shlex45from tui.widgets.log_pane import LogPane6from tui.widgets.sync_plan_pane import SyncPlanPane78if TYPE_CHECKING:9    from tui.model.sync_plan_model import SyncPlanModel

1def __init__(2    self,3    *,4    queue_path: Path | None = None,5    ascii_only: bool = False,6    poll_interval: float = 2.0,7    plan_path: Path | None = None,8    adk_bin: Path | None = None,9) -> None:10    super().__init__()11    self._queue_path = queue_path12    self._ascii_only = ascii_only13    self.poll_interval = poll_interval14    self._plan_path = plan_path15    self._adk_bin = adk_bin  # Path to `adk` binary; None → resolve to repo's bin/adk16    self._filter_mode: FilterMode = "all"17    self._sort_mode: SortMode = "fifo"18    self._model: QueueModel | None = None19    self._plan_model: SyncPlanModel | None = None20    self._rows_by_url: dict[str, QueueRow] = {}21    self._sync_proc: asyncio.subprocess.Process | None = None22    self._sync_task: asyncio.Task | None = None

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

1Binding("s", "sync", "sync"),

1async def on_mount(self) -> None:2    from tui.model.queue_model import QueueModel3    from tui.model.sync_plan_model import SyncPlanModel45    self._model = QueueModel(queue_path=self._queue_path)6    self._plan_model = SyncPlanModel(plan_path=self._plan_path)7    self.query_one(FooterBar).update_status(self._filter_mode, self._sort_mode, sync_running=False)8    self._reload(force=True)9    self._reload_plan(force=True)10    self.set_interval(self.poll_interval, self._maybe_reload)

1def _reload_plan(self, *, force: bool = False) -> None:2    if self._plan_model is None:3        return4    if not force and not self._plan_model.has_changed():5        return6    snapshot = self._plan_model.snapshot()7    self.query_one(SyncPlanPane).update_snapshot(snapshot, 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()

1def action_sync(self) -> None:2    if self._sync_proc is not None and self._sync_proc.returncode is None:3        self.query_one(LogPane).announce("(sync already running — wait or quit and restart)")4        return5    self._sync_task = asyncio.create_task(self._run_sync())67async def _run_sync(self) -> None:8    log_pane = self.query_one(LogPane)9    queue_arg: list[str] = []10    if self._queue_path is not None:11        queue_arg = ["--queue", str(self._queue_path)]12    adk = self._resolve_adk_bin()13    if adk is None:14        log_pane.announce("(error: could not locate `adk` binary)")15        return16    cmd = [str(adk), "pr-sync", *queue_arg]17    log_pane.announce(f"$ {' '.join(shlex.quote(c) for c in cmd)}")18    self.query_one(FooterBar).update_status(self._filter_mode, self._sort_mode, sync_running=True)19    env = dict(os.environ)20    if self._plan_path is not None:21        env["ADK_TUI_PLAN_PATH"] = str(self._plan_path)22    try:23        self._sync_proc = await asyncio.create_subprocess_exec(24            *cmd,25            stdout=asyncio.subprocess.PIPE,26            stderr=asyncio.subprocess.STDOUT,27            env=env,28        )29    except (FileNotFoundError, PermissionError) as exc:30        log_pane.announce(f"(error: {exc})")31        self._sync_proc = None32        self.query_one(FooterBar).update_status(self._filter_mode, self._sort_mode, sync_running=False)33        return34    assert self._sync_proc.stdout is not None35    while True:36        line = await self._sync_proc.stdout.readline()37        if not line:38            break39        log_pane.write(line.decode(errors="replace").rstrip("\n"))40    rc = await self._sync_proc.wait()41    log_pane.announce(f"(pr-sync exited rc={rc})")42    self._sync_proc = None43    self.query_one(FooterBar).update_status(self._filter_mode, self._sort_mode, sync_running=False)44    self._reload_plan(force=True)45    if self._model is not None:46        # The sync rewrote the queue; force a re-read.47        self._reload(force=True)4849def _resolve_adk_bin(self) -> Path | None:50    if self._adk_bin is not None:51        return self._adk_bin52    # Fall back to repo's bin/adk53    repo_root = Path(__file__).resolve().parent.parent54    candidate = repo_root / "bin" / "adk"55    if candidate.exists():56        return candidate57    # Last-resort: hope it's on PATH58    return Path("adk")

1async def on_unmount(self) -> None:2    if self._sync_proc is not None and self._sync_proc.returncode is None:3        try:4            self._sync_proc.terminate()5            try:6                await asyncio.wait_for(self._sync_proc.wait(), timeout=2.0)7            except asyncio.TimeoutError:8                self._sync_proc.kill()9        except ProcessLookupError:10            pass

1def update_status(self, filter_mode: str, sort_mode: str, *, sync_running: bool = False) -> None:2    sync_label = "[s] sync (running…)" if sync_running else "[s] sync"3    text = (4        f"[?] help  [f] filter:{filter_mode}  [S] sort:{sort_mode}"5        f"  [j/k] nav  [q] quit  ·  {sync_label}  [r] run (disabled)"6    )7    self.update(text)

1adk · queue: 12 · ready: 8 · in-review: 2/4 · platform: github+bitbucket2┌─ Queue ──────────────────────┐┌─ Details ─────────────────────────────┐3│ ✓  1234  acme/sf-bff …       ││ acme/sf-bff#1234                       │4│ ⚙  1235  acme/sf-bff …       ││ Title: …                                │5│ …                            ││ Author: …                               │6└──────────────────────────────┘└────────────────────────────────────────┘7Sync plan (running · 3/8 steps)8  ✓  pr-scan9  ✓  pr-queue update --all10  ⚡ pr-queue clean (merged)11  …  pr-task clean-orphans12  …  pr-queue remind13  …  base-index audit14  …  auto-base cleanup15  …  pr-task prepare --all16┌─ Log ─────────────────────────────────────────────────────────────────┐17│ $ /Users/.../bin/adk pr-sync                                          │18│ pr-scan: scanning channel #eng-prs since 1.0h ago                     │19│ pr-scan: found 3 new PR URLs                                          │20│ ...                                                                    │21└────────────────────────────────────────────────────────────────────────┘22[?] help  [f] filter:all  [S] sort:fifo  [j/k] nav  [q] quit  ·  [s] sync (running…)  [r] run (disabled)

1@pytest.fixture2def fake_plan_path(tmp_path: Path) -> Path:3    return tmp_path / "sync-plan.json"456@pytest.fixture7def sync_plan_in_progress(tmp_path: Path) -> Path:8    """A plan file with 2 ok steps + 1 running step + 5 pending."""9    p = tmp_path / "sync-plan.json"10    p.write_text(json.dumps({11        "version": 1,12        "queue": "/tmp/q.json5",13        "argv": [],14        "started_at": "2026-05-22T14:00:00Z",15        "updated_at": "2026-05-22T14:01:30Z",16        "completed_at": None,17        "rc": None,18        "steps": [19            {"name": "pr-scan", "status": "ok", "rc": 0, "started_at": "...", "completed_at": "..."},20            {"name": "pr-queue update --all", "status": "ok", "rc": 0, "started_at": "...", "completed_at": "..."},21            {"name": "pr-queue clean (merged)", "status": "running", "rc": None, "started_at": "...", "completed_at": None},22            *[{"name": n, "status": "pending", "rc": None, "started_at": None, "completed_at": None}23              for n in ["pr-task clean-orphans", "pr-queue remind", "base-index audit", "auto-base cleanup", "pr-task prepare --all"]],24        ],25    }))26    return p272829@pytest.fixture30def fake_adk_script(tmp_path: Path) -> Path:31    p = tmp_path / "fake-adk"32    p.write_text("#!/bin/sh\n"33                 "echo 'pr-scan: running'\n"34                 "echo 'pr-scan: 0 new'\n"35                 "echo 'done'\n"36                 "exit 0\n")37    p.chmod(0o755)38    return p