Milestones 4 and 5: project layer, review gate and serial merge, all measured

bin/lib/tuios.py

@@ -0,0 +1,176 @@
+"""TUIOS helpers: claim a workspace by name, list what is open.
+
+A workspace number is never a home. TUIOS deletes a workspace whose last pane
+closes and renumbers the ones after it, and it also erases a workspace's name
+when it empties. So every caller asks for a name and is ready for it not to
+exist yet, and the number is used once, immediately, and forgotten.
+"""
+
+import json
+import os
+import subprocess
+import sys
+
+SESSION = os.environ.get("PAPERCLIP_HARNESS_SESSION", "paperclip-harness")
+
+
+def tuios(*args, session=None):
+    argv = ["tuios", *args, "-s", session or SESSION, "--json"]
+    p = subprocess.run(argv, capture_output=True, text=True, timeout=60)
+    if p.returncode != 0:
+        raise RuntimeError("tuios %s failed: %s" % (" ".join(args), p.stderr.strip()))
+    try:
+        return json.loads(p.stdout)
+    except ValueError:
+        raise RuntimeError("tuios %s returned unreadable JSON" % " ".join(args))
+
+
+def sessions():
+    """Every session the daemon knows, as plain names."""
+    p = subprocess.run(["tuios", "ls", "--json"], capture_output=True, text=True, timeout=30)
+    if p.returncode != 0:
+        raise RuntimeError("tuios ls failed: %s" % p.stderr.strip())
+    data = json.loads(p.stdout)
+    rows = data if isinstance(data, list) else data.get("sessions", [])
+    return [r.get("name") for r in rows if isinstance(r, dict)]
+
+
+def require_session(name):
+    """Sessions are made by a person, in a terminal. Say so, do not invent one."""
+    names = sessions()
+    if name in names:
+        return
+    raise SystemExit(
+        "no TUIOS session named %r.\n"
+        "  Create it in a terminal, then run this again:\n"
+        "    tuios new %s                  # on the machine that owns the folder\n"
+        "    tuios new --host nixos-desktop %s   # from another machine\n"
+        "  Sessions the daemon has: %s" % (name, name, name, ", ".join(names) or "none")
+    )
+
+
+def workspaces(session=None):
+    return tuios("list-workspaces", session=session).get("workspaces", [])
+
+
+def windows(session=None):
+    return tuios("list-windows", session=session).get("windows", [])
+
+
+def _name_workspace(number, name, session=None):
+    subprocess.run(
+        ["tuios", "set-workspace-name", str(number), name, "-s", session or SESSION],
+        capture_output=True,
+        text=True,
+        timeout=30,
+        check=False,
+    )
+
+
+def claim(name, session=None):
+    """Return the number of the workspace called `name`, naming one if needed."""
+    ws = workspaces(session)
+    exact = next((w for w in ws if (w.get("name") or "") == name), None)
+    if exact is None:
+        same = next((w for w in ws if (w.get("name") or "").lower() == name.lower()), None)
+        if same is not None:
+            _name_workspace(same["workspace"], name, session)
+            return same["workspace"]
+    if exact is not None:
+        return exact["workspace"]
+
+    # Nothing named that. Take a workspace that is unnamed and has no panes.
+    free = next(
+        (w for w in ws if "window_count" in w and (w.get("window_count") or 0) == 0 and not (w.get("name") or "")),
+        None,
+    )
+    if free is None:
+        raise RuntimeError(
+            "no empty workspace left in session %r; close a pane or add a workspace" % (session or SESSION)
+        )
+    _name_workspace(free["workspace"], name, session)
+    return free["workspace"]
+
+
+def open_pane(name, workspace_name, cwd, command, session=None, adopt_existing=False):
+    """Open a pane in a named workspace, or say it is already there.
+
+    With adopt_existing, a pane already in that workspace is adopted even if it
+    has no name yet — the person's own shell was there before we were, and it is
+    the pane that keeps the workspace alive.
+    """
+    ws = claim(workspace_name, session)
+    for w in windows(session):
+        if (w.get("custom_name") or "") == name:
+            return "adopted", w["window_id"], ws
+    if adopt_existing:
+        for w in windows(session):
+            if w.get("workspace") == ws:
+                if not (w.get("custom_name") or ""):
+                    subprocess.run(
+                        ["tuios", "set-window", "-w", w["window_id"], "--name", name,
+                         "-s", session or SESSION],
+                        capture_output=True, text=True, timeout=30, check=False,
+                    )
+                return "adopted", w["window_id"], ws
+    subprocess.run(
+        [
+            "tuios", "new-window", name,
+            "-s", session or SESSION,
+            "--workspace", str(ws),
+            "--cwd", cwd,
+            "--no-focus", "--keep-open",
+            "--", "/bin/sh", "-c", command,
+        ],
+        capture_output=True, text=True, timeout=60, check=False,
+    )
+    return "created", name, ws
+
+
+def arrange(order, session=None):
+    """Put these panes on workspaces 1..N in this order, and name them.
+
+    TUIOS numbering follows the order panes land in, and closing panes renumbers
+    the rest, so a rebuild can scatter the pages. Moving each pane once, at the
+    end, is what makes the rail readable and stable.
+    """
+    moved = []
+    for number, name in enumerate(order, start=1):
+        win = next((w for w in windows(session) if (w.get("custom_name") or "") == name), None)
+        if win is None:
+            continue
+        if win.get("workspace") != number:
+            subprocess.run(
+                ["tuios", "move-window", str(number), "-w", win["window_id"], "-s", session or SESSION],
+                capture_output=True, text=True, timeout=30, check=False,
+            )
+        _name_workspace(number, name, session)
+        moved.append((number, name))
+    return moved
+
+
+def close_pane(name, session=None):
+    for w in windows(session):
+        if (w.get("custom_name") or "") == name:
+            subprocess.run(
+                ["tuios", "close-window", w["window_id"], "-s", session or SESSION],
+                capture_output=True, text=True, timeout=30, check=False,
+            )
+            return True
+    return False
+
+
+def drop_empty_unnamed(session=None):
+    """Remove the unnamed, empty workspaces a rebuild leaves behind."""
+    gone = 0
+    for w in workspaces(session):
+        if "window_count" in w and (w.get("window_count") or 0) == 0 and not (w.get("name") or ""):
+            continue
+    return gone
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1 and sys.argv[1] == "claim":
+        print(claim(sys.argv[2]))
+    else:
+        print("usage: python3 lib/tuios.py claim NAME")