TUI sub-phase η — binding interface SPEC

1from __future__ import annotations23from textual.app import ComposeResult4from textual.binding import Binding5from textual.containers import Container6from textual.screen import ModalScreen7from textual.widgets import Input, Static8910class PromptScreen(ModalScreen[str | None]):11    """Single-input modal. dismiss() with the typed value on submit,12    dismiss(None) on cancel (escape)."""1314    BINDINGS = [Binding("escape", "cancel", show=False)]1516    DEFAULT_CSS = """17    PromptScreen {18        align: center middle;19    }20    PromptScreen > Container {21        width: 60;22        height: auto;23        padding: 1 2;24        border: round $accent;25        background: $surface;26    }27    PromptScreen Static {28        width: 100%;29    }30    PromptScreen Input {31        width: 100%;32    }33    """3435    def __init__(self, label: str, placeholder: str = "") -> None:36        super().__init__()37        self._label = label38        self._placeholder = placeholder3940    def compose(self) -> ComposeResult:41        with Container():42            yield Static(self._label, markup=False)43            yield Input(placeholder=self._placeholder)4445    def on_input_submitted(self, event: Input.Submitted) -> None:46        self.dismiss(event.value)4748    def action_cancel(self) -> None:49        self.dismiss(None)

1value = await self.push_screen_wait(PromptScreen("Add PR", "URL, owner/repo#N, or number"))2if not value:3    return

1Binding("plus", "add_pr", "add-pr"),2Binding("b", "repos", "repos"),

1async def action_add_pr(self) -> None:2    from tui.screens.prompt_screen import PromptScreen3    value = await self.push_screen_wait(4        PromptScreen("Add PR", "URL, owner/repo#N, or PR number")5    )6    if not value:7        return8    busy = self._busy_label()9    if busy is not None:10        self.query_one(LogPane).write(f"(can't add PR — {busy} already running)")11        return12    self._add_pr_task = asyncio.create_task(self._run_add_pr(value.strip()))

1async def _run_add_pr(self, text: str) -> None:2    log_pane = self.query_one(LogPane)3    adk = self._resolve_adk_bin()4    cmd: list[str] = [str(adk), "pr-queue", "add", text, "-y"]5    if self._queue_path is not None:6        cmd += ["--queue", str(self._queue_path)]7    log_pane.write(f"$ {' '.join(shlex.quote(c) for c in cmd)}")8    try:9        proc = await asyncio.create_subprocess_exec(10            *cmd,11            stdout=asyncio.subprocess.PIPE,12            stderr=asyncio.subprocess.STDOUT,13        )14    except (FileNotFoundError, PermissionError, OSError) as exc:15        log_pane.write(f"(error: {exc})")16        return17    assert proc.stdout is not None18    while True:19        line = await proc.stdout.readline()20        if not line:21            break22        log_pane.write(line.decode(errors="replace").rstrip("\n"))23    rc = await proc.wait()24    log_pane.write(f"(pr-queue add exited rc={rc})")25    if self._model is not None:26        self._reload(force=True)

1def action_repos(self) -> None:2    from tui.screens.repo_screen import RepoScreen3    self.push_screen(RepoScreen(4        repos_dir=self._repos_dir,5        adk_bin_resolver=self._resolve_adk_bin,6    ))

1def _busy_label(self) -> str | None:2    if self._sync_proc is not None and self._sync_proc.returncode is None:3        return "sync"4    if self._batch_task is not None and not self._batch_task.done():5        return "batch"6    if self._add_pr_task is not None and not self._add_pr_task.done():7        return "add-pr"8    if self._review_workers:9        return "review"10    return None

1from __future__ import annotations23import asyncio4import shlex5from pathlib import Path6from typing import Callable78from textual.app import ComposeResult9from textual.binding import Binding10from textual.containers import Container, Vertical11from textual.screen import Screen12from textual.widgets import Static, Tree1314from tui.model.repo_model import RepoModel, RepoRow, RepoBranchRow15from tui.screens.prompt_screen import PromptScreen161718class RepoScreen(Screen):19    BINDINGS = [20        Binding("escape", "back", "back"),21        Binding("plus", "add_repo", "add-repo"),22        Binding("a", "add_branch", "add-branch"),23        Binding("R", "rebuild", "rebuild"),24        Binding("d", "delete_auto_base", "delete-auto-base"),25        Binding("j", "cursor_down", show=False),26        Binding("k", "cursor_up", show=False),27    ]2829    DEFAULT_CSS = """30    RepoScreen { layout: vertical; }31    RepoScreen Static#repo_header {32        height: 1;33        background: $primary-background;34        color: $text;35        padding: 0 1;36    }37    RepoScreen Tree {38        height: 1fr;39    }40    RepoScreen Static#repo_footer {41        dock: bottom;42        height: 1;43        background: $primary-background;44        color: $text;45        padding: 0 1;46    }47    """4849    def __init__(50        self,51        *,52        repos_dir: Path | None = None,53        adk_bin_resolver: Callable[[], Path] | None = None,54        poll_interval: float = 5.0,55    ) -> None:56        super().__init__()57        self._model = RepoModel(repos_dir=repos_dir)58        self._adk_bin_resolver = adk_bin_resolver or (lambda: Path("adk"))59        self._poll_interval = poll_interval60        self._action_task: asyncio.Task | None = None6162    def compose(self) -> ComposeResult:63        yield Static("Repos · [escape] back  [+] add  [a] add-branch  [R] rebuild  [d] delete-auto-base",64                     id="repo_header", markup=False)65        yield Tree("Repos", id="repo_tree")66        yield Static("", id="repo_footer", markup=False)6768    def on_mount(self) -> None:69        self._refresh()70        self.set_interval(self._poll_interval, self._refresh)7172    def action_back(self) -> None:73        self.app.pop_screen()7475    def action_cursor_down(self) -> None:76        self.query_one(Tree).action_cursor_down()7778    def action_cursor_up(self) -> None:79        self.query_one(Tree).action_cursor_up()8081    def _refresh(self) -> None:82        rows = self._model.snapshot()83        tree = self.query_one(Tree)84        tree.clear()85        for repo in rows:86            node = tree.root.add(_format_repo_label(repo), expand=True)87            for br in repo.branches:88                node.add(_format_branch_label(br), data=("branch", repo.name, br.branch))89            node.data = ("repo", repo.name, None)90        self.query_one("#repo_footer", Static).update(91            f"{len(rows)} repos · {sum(len(r.branches) for r in rows)} branches"92        )9394    # --- actions ---95    async def action_add_repo(self) -> None:96        url = await self.app.push_screen_wait(97            PromptScreen("Add repo", "git URL (git@... or https://...)")98        )99        if not url:100            return101        await self._spawn_subprocess(102            [str(self._adk_bin_resolver()), "repo", "add", url.strip(), "-y"],103            label="repo-add",104        )105106    async def action_add_branch(self) -> None:107        # Use the highlighted repo (if a branch row is highlighted, use its parent).108        repo_name = self._cursor_repo_name()109        if not repo_name:110            self._log_screen("(no repo highlighted)")111            return112        branch = await self.app.push_screen_wait(113            PromptScreen(f"Add branch to {repo_name}", "branch name (e.g., release/v2.1)")114        )115        if not branch:116            return117        await self._spawn_subprocess(118            [str(self._adk_bin_resolver()), "repo", "branch", "add",119             repo_name, "--branch", branch.strip(), "-y"],120            label="branch-add",121        )122123    async def action_rebuild(self) -> None:124        ctx = self._cursor_context()125        if ctx is None:126            self._log_screen("(no row highlighted)")127            return128        kind, repo_name, branch = ctx129        cmd = [str(self._adk_bin_resolver()), "repo", "rebuild-index", repo_name]130        if kind == "branch" and branch:131            cmd += ["--branch", branch]132        cmd += ["-y"]133        await self._spawn_subprocess(cmd, label="rebuild-index")134135    async def action_delete_auto_base(self) -> None:136        ctx = self._cursor_context()137        if ctx is None or ctx[0] != "branch":138            self._log_screen("(highlight an auto-base branch to delete)")139            return140        _, repo_name, branch = ctx141        await self._spawn_subprocess(142            [str(self._adk_bin_resolver()), "repo", "auto-bases", "clean",143             "--force", repo_name, "--branch", branch, "-y"],144            label="auto-base-clean",145        )146147    # --- helpers ---148    def _cursor_context(self) -> tuple[str, str, str | None] | None:149        tree = self.query_one(Tree)150        node = tree.cursor_node151        if node is None or node.data is None:152            return None153        return node.data  # type: ignore[return-value]154155    def _cursor_repo_name(self) -> str | None:156        ctx = self._cursor_context()157        if ctx is None:158            return None159        return ctx[1]160161    def _log_screen(self, msg: str) -> None:162        self.query_one("#repo_footer", Static).update(msg)163164    async def _spawn_subprocess(self, cmd: list[str], *, label: str) -> None:165        self._log_screen(f"$ {' '.join(shlex.quote(c) for c in cmd)}")166        try:167            proc = await asyncio.create_subprocess_exec(168                *cmd,169                stdout=asyncio.subprocess.PIPE,170                stderr=asyncio.subprocess.STDOUT,171            )172        except (FileNotFoundError, PermissionError, OSError) as exc:173            self._log_screen(f"(error: {exc})")174            return175        # Drain output line-by-line to the footer; the user sees the last line.176        # For richer logging, we could route to the main screen's LogPane, but177        # that requires app-level coordination; footer is good enough for now.178        assert proc.stdout is not None179        last_line = ""180        while True:181            line = await proc.stdout.readline()182            if not line:183                break184            last_line = line.decode(errors="replace").rstrip("\n")185            self._log_screen(last_line[:200])186        rc = await proc.wait()187        self._log_screen(f"{label}: rc={rc}  ·  {last_line[:150]}")188        self._refresh()189190191def _format_repo_label(repo: RepoRow) -> str:192    return f"{repo.name}  ({repo.url})"193194195def _format_branch_label(br: RepoBranchRow) -> str:196    origin = "user" if br.created_by == "user" else "auto"197    age = _format_age_s(br.age_s)198    indexed = br.last_indexed_at or "—"199    return f"{br.branch}  ·  {origin}  ·  indexed {indexed}  ·  used {age}"200201202def _format_age_s(seconds: float | None) -> str:203    if seconds is None:204        return "—"205    if seconds < 60:206        return f"{int(seconds)}s ago"207    minutes = int(seconds // 60)208    if minutes < 60:209        return f"{minutes}m ago"210    hours = int(minutes // 60)211    if hours < 24:212        return f"{hours}h ago"213    return f"{int(hours // 24)}d ago"

1from .prompt_screen import PromptScreen2from .repo_screen import RepoScreen34__all__ = ["PromptScreen", "RepoScreen"]

1  +              add PR (modal — URL, owner/repo#N, or number)2  b              switch to repos screen (manage repos + branches)

1from __future__ import annotations23import json4from dataclasses import dataclass, field5from datetime import datetime, timezone6from pathlib import Path7from typing import Callable8910@dataclass(frozen=True)11class RepoBranchRow:12    repo_name: str13    branch: str14    slug: str15    created_by: str  # "user" | "auto"16    last_indexed_at: str | None17    last_used_at: str | None18    age_s: float | None  # seconds since last_used_at, or None if not set19    auto_reason: str | None202122@dataclass(frozen=True)23class RepoRow:24    name: str25    url: str26    default_branch: str27    branches: tuple[RepoBranchRow, ...]282930def default_repos_dir() -> Path:31    return Path.home() / ".agents-devkit" / "repos"323334def _parse_iso(ts: str | None) -> datetime | None:35    if not ts:36        return None37    try:38        if ts.endswith("Z"):39            ts = ts[:-1] + "+00:00"40        return datetime.fromisoformat(ts)41    except ValueError:42        return None434445class RepoModel:46    """Reads ~/.agents-devkit/repos/ and exposes repos + branches as47    frozen dataclasses. Mtime-gated via a directory-fingerprint signature."""4849    def __init__(50        self,51        repos_dir: Path | None = None,52        *,53        now_fn: Callable[[], datetime] | None = None,54    ) -> None:55        self.repos_dir = repos_dir if repos_dir is not None else default_repos_dir()56        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        if not self.repos_dir.exists():63            return ()64        items: list[tuple[str, float]] = []65        try:66            for p in self.repos_dir.iterdir():67                if not p.is_dir():68                    continue69                meta = p / "repo-meta.json"70                try:71                    items.append((p.name, meta.stat().st_mtime if meta.exists() else 0.0))72                except OSError:73                    items.append((p.name, 0.0))74                # also sample each branch-meta.json mtime for change detection75                for child in p.iterdir():76                    if child.is_dir() and child.name.startswith("branch-"):77                        bm = child / "branch-meta.json"78                        try:79                            items.append((f"{p.name}/{child.name}",80                                          bm.stat().st_mtime if bm.exists() else 0.0))81                        except OSError:82                            items.append((f"{p.name}/{child.name}", 0.0))83        except OSError:84            return ()85        items.sort()86        return tuple(items)8788    def has_changed(self) -> bool:89        cur = self._signature()90        if cur != self._last_signature:91            return True92        return False9394    def snapshot(self) -> list[RepoRow]:95        self._last_signature = self._signature()96        if not self.repos_dir.exists():97            return []98        now = self._now_fn()99        rows: list[RepoRow] = []100        for p in sorted(self.repos_dir.iterdir()):101            if not p.is_dir():102                continue103            meta_path = p / "repo-meta.json"104            if not meta_path.exists():105                continue106            try:107                raw = json.loads(meta_path.read_text())108            except (OSError, json.JSONDecodeError):109                continue110            if not isinstance(raw, dict):111                continue112            branches = self._collect_branches(p, raw.get("name") or p.name, now)113            rows.append(RepoRow(114                name=str(raw.get("name") or p.name),115                url=str(raw.get("url") or raw.get("input_url") or ""),116                default_branch=str(raw.get("default_branch") or ""),117                branches=tuple(branches),118            ))119        return rows120121    def _collect_branches(self, repo_dir: Path, repo_name: str,122                          now: datetime) -> list[RepoBranchRow]:123        out: list[RepoBranchRow] = []124        for child in sorted(repo_dir.iterdir()):125            if not (child.is_dir() and child.name.startswith("branch-")):126                continue127            bm = child / "branch-meta.json"128            if not bm.exists():129                continue130            try:131                raw = json.loads(bm.read_text())132            except (OSError, json.JSONDecodeError):133                continue134            if not isinstance(raw, dict):135                continue136            last_used = _parse_iso(raw.get("last_used_at"))137            age_s = (now - last_used).total_seconds() if last_used is not None else None138            out.append(RepoBranchRow(139                repo_name=repo_name,140                branch=str(raw.get("branch") or ""),141                slug=str(raw.get("slug") or child.name.removeprefix("branch-")),142                created_by=str(raw.get("created_by") or "user"),143                last_indexed_at=raw.get("last_indexed_at"),144                last_used_at=raw.get("last_used_at"),145                age_s=age_s,146                auto_reason=raw.get("auto_reason"),147            ))148        return out

1@pytest.fixture2def fake_repos_dir(tmp_path: Path) -> Path:3    import json4    root = tmp_path / "repos"5    repo = root / "fake-repo"6    repo.mkdir(parents=True)7    (repo / "repo-meta.json").write_text(json.dumps({8        "name": "fake-repo", "url": "git@github.com:acme/fake.git",9        "default_branch": "main",10    }))11    branch = repo / "branch-main"12    branch.mkdir()13    (branch / "branch-meta.json").write_text(json.dumps({14        "branch": "main", "slug": "main", "created_by": "user",15        "last_indexed_at": "2026-05-22T10:00:00Z",16        "last_used_at": "2026-05-22T13:00:00Z",17    }))18    auto_branch = repo / "branch-feat-x"19    auto_branch.mkdir()20    (auto_branch / "branch-meta.json").write_text(json.dumps({21        "branch": "feat/x", "slug": "feat-x", "created_by": "auto",22        "last_indexed_at": "2026-05-22T13:30:00Z",23        "last_used_at": "2026-05-22T13:30:00Z",24        "auto_reason": "3 PRs target feat/x",25    }))26    return root