"""cmux command helpers for okstra-owned worker surfaces.

The tmux sibling of this module is `tmux.py`. cmux is a GUI app, so there is no
detached-server equivalent here — every surface okstra creates lands in the
workspace the user is already looking at.
"""
from __future__ import annotations

import json
import os
import shlex
import shutil
import socket
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Collection, Sequence

PING_OK = "PONG"

# cmux answers over a local unix socket, so a probe that has not returned in a
# few seconds means the app is wedged or gone — not that it is being slow.
PROBE_TIMEOUT_SECONDS = 5

# A login shell sources the user's whole rc chain, which can be slow on a
# developer machine, so this gets far more room than a socket probe.
LOGIN_SHELL_TIMEOUT_SECONDS = 15

# cmux plants a per-surface directory of CLI shims on PATH; every entry under it
# re-enters cmux's own agent wrapper instead of the real CLI.
SHIM_DIR_MARKER = "cmux-cli-shims"

# The first three workers share one column, so the lead keeps two fifths. A
# fourth worker opens a second column; the lead drops to two sixths so that
# extra column has room. Splitting the first column sideways before it is full
# would make the second worker narrower than the first.
WORKERS_PER_COLUMN = 3
LEAD_SHARE_WITH_WORKERS = 2 / 5
LEAD_SHARE_WITH_MULTIPLE_COLUMNS = 2 / 6

# Three fifths is what a worker needs to be worth watching: measured against a
# Claude Code worker, 67 columns renders losslessly and 33 drops content off the
# right edge, and three fifths clears 67 on any window wide enough to hold two
# panes at all.
#
# With no workers on screen the lead has nothing to share with, so it takes the
# whole workspace back rather than sitting at its working width.
LEAD_SHARE_ALONE = 1.0

# Two passes per pane above the bottom. The first column holds at most three
# workers; eight moves is that bound with room to spare.
MAX_WORKER_HEIGHT_MOVES = 8

# Sidebar entries are keyed by source so tools do not overwrite each other's.
SIDEBAR_SOURCE = "okstra"

# Why a run that prepare recorded as cmux can no longer see cmux. Kept apart
# because they call for opposite responses: a sanitized environment or a denied
# socket means a sandbox stands in the way and the fallback is doomed with it,
# while a quit app leaves the worker CLIs perfectly able to run.
LOST_NOTHING = ""
LOST_ENVIRONMENT = "environment"
LOST_DENIED = "denied"
LOST_GONE = "gone"

# Verdicts from `socket_reachability`. `denied` is the one that matters: it
# means a sandbox stands between this process and cmux, and the same sandbox
# hides the worker CLIs' own config, so falling back is already doomed.
SOCKET_OK = "ok"
SOCKET_DENIED = "denied"
SOCKET_MISSING = "missing"
SOCKET_UNREACHABLE = "unreachable"

SOCKET_PROBE_TIMEOUT_SECONDS = 2


def cmux_cli_path() -> str:
    """Absolute path to the cmux CLI, or "" when cmux is not installed.

    The bundled CLI wins over PATH: cmux documents the /usr/local/bin/cmux
    symlink as a manual step, so a working install may leave PATH untouched.
    """
    bundled = os.environ.get("CMUX_BUNDLED_CLI_PATH", "")
    if bundled and os.access(bundled, os.X_OK):
        return bundled
    return shutil.which("cmux") or ""


def run_cmux(
    args: Sequence[str], *, timeout: int = PROBE_TIMEOUT_SECONDS
) -> subprocess.CompletedProcess[str]:
    cli = cmux_cli_path()
    if not cli:
        raise FileNotFoundError(
            "cmux CLI not found: CMUX_BUNDLED_CLI_PATH unset and no cmux on PATH"
        )
    return subprocess.run(
        [cli, *args],
        capture_output=True,
        text=True,
        timeout=timeout,
        check=False,
    )


def cmux_available() -> bool:
    """True only when okstra can drive cmux *and* knows where the lead sits.

    Being able to drive cmux is not sufficient. Workers attach to the lead's
    workspace, so a run whose lead location is unresolvable would open panes on
    a screen nobody is watching; degrading to the blocking wrapper is the better
    failure. Nested tmux is exactly that case — see `resolve_lead_workspace`.
    """
    if not cmux_cli_path():
        return False
    # Cheapest discriminator first: no workspace in the environment means this
    # is not a cmux-hosted session, and the probes below would only confirm that
    # the app happens to be installed.
    if not _lead_workspace_env():
        return False
    if not _ping_answers():
        return False
    return bool(resolve_lead_workspace())


def resolve_lead_workspace() -> str:
    """The lead's workspace UUID, or "" when it cannot be resolved.

    The UUID is read from the environment rather than from `identify`, whose
    refs are positional and shift as surfaces open and close. `identify` is
    consulted only to prove the environment is current: a tmux server freezes
    the CMUX_* block it was launched with, so those variables can outlive the
    surface they name.
    """
    workspace = _lead_workspace_env()
    if not workspace:
        return ""
    if not identify_caller():
        return ""
    return workspace


def _lead_workspace_env() -> str:
    return os.environ.get("CMUX_WORKSPACE_ID", "").strip()


@dataclass(frozen=True)
class PaneGeometry:
    """One pane as `pane.list` reports it.

    `columns` and `rows` come straight from cmux rather than being derived from
    the container frame, so a display with a different cell size needs no
    conversion here.
    """

    pane_id: str
    surface_ids: tuple[str, ...]
    columns: int
    rows: int
    x: int
    y: int
    cell_width_points: int
    width_points: float = 0.0
    ref: str = ""
    selected_surface_id: str = ""
    height_points: float = 0.0


@dataclass(frozen=True)
class Placement:
    """Where the next worker goes: split `pane_id` in `direction`."""

    pane_id: str
    direction: str


@dataclass(frozen=True)
class WorkerHeightResize:
    """One vertical border move that equalizes the worker column.

    `down` on a pane grows it into the neighbour below. `up` on a pane pulls
    the shared border up, shrinking the neighbour above — the top pane has no
    upper border, so shrinking it is the pane below's request.
    """

    pane_id: str
    direction: str
    amount: int


def plan_worker_placement(
    panes: Sequence[PaneGeometry],
    *,
    lead_pane_id: str,
    owned_surface_ids: Collection[str],
) -> Placement:
    """Pick the next worker slot from the workspace's current geometry.

    The first worker splits off the lead to the right. The next two fill that
    column downward from the bottom, so three workers share one width. From the
    fourth onward the new pane splits worker n-3 to the right — the matching
    row of the previous column — which is column-major order of the current
    panes sorted by (x, y).

    Stateless by design: okstra records surface UUIDs, never a layout, so a
    resumed or crashed run cannot carry a layout model that no longer matches
    the screen. Every dispatch re-reads the panes and derives the next slot.

    Those same UUIDs say which panes okstra may place into. A workspace also
    holds panes okstra never opened — another agent session, a shell the user
    keeps around — and "not the lead" does not make a pane a worker slot. Taken
    as one, a stranger's pane is split or stacked into: the workers land as
    background tabs in someone else's window, so nothing appears on screen and
    that window grows tabs it did not ask for.
    """
    workers = [
        pane
        for pane in panes
        if pane.pane_id != lead_pane_id
        and _holds_an_okstra_surface(pane, owned_surface_ids)
    ]
    if not workers:
        return Placement(pane_id=lead_pane_id, direction="right")
    if len(workers) < WORKERS_PER_COLUMN:
        return _extend_the_worker_column(workers)
    return _extend_the_worker_grid(workers)


def lead_target_width(
    panes: Sequence[PaneGeometry],
    *,
    lead_pane_id: str,
    owned_surface_ids: Collection[str],
    container_width_points: float,
) -> float:
    """How wide the lead should be, in the points `pixel_frame` reports.

    Two fifths while the first worker column is filling, two sixths once a
    fourth worker is on screen, all of it when they are gone. The share is
    taken of what okstra may actually place into, not of the window: a
    workspace can hold panes okstra never opened, and handing the lead the
    whole container would shove those off their own width. Their width is
    subtracted first and the share applies to the remainder.

    Deriving the target from the container rather than from a fixed column count
    is what keeps the split honest at any window size — a hardcoded 80 columns
    is two fifths of one particular display and an arbitrary slice of every
    other.
    """
    strangers = sum(
        pane.width_points
        for pane in panes
        if pane.pane_id != lead_pane_id
        and not _holds_an_okstra_surface(pane, owned_surface_ids)
    )
    usable = container_width_points - strangers
    if usable <= 0:
        return 0.0
    worker_count = sum(
        1
        for pane in panes
        if pane.pane_id != lead_pane_id
        and _holds_an_okstra_surface(pane, owned_surface_ids)
    )
    if worker_count == 0:
        share = LEAD_SHARE_ALONE
    elif worker_count <= WORKERS_PER_COLUMN:
        share = LEAD_SHARE_WITH_WORKERS
    else:
        share = LEAD_SHARE_WITH_MULTIPLE_COLUMNS
    return usable * share


def lead_resize_points(lead: PaneGeometry, *, target_width_points: float) -> int:
    """How far to move the lead's right border, in the points `pane.resize` takes.

    Signed: positive when the lead is too wide and the border comes in, negative
    when it is too narrow and the border goes out.

    Both directions are needed. A split halves whatever pane it lands on, and
    the first worker of every round lands on the lead — so a rule that only ever
    shrinks leaves that half permanent, and the round after it takes half of
    what is left. Measured on this display: 215 columns becomes 107, then 53,
    then 26, until neither the lead nor its workers can be read.

    Measured in the same points `pixel_frame` reports, so no cell-size
    conversion happens here at all. A column count would need one, and would
    round the target to a whole cell before the border ever moved.
    """
    return round(lead.width_points - target_width_points)


def _holds_an_okstra_surface(
    pane: PaneGeometry, owned_surface_ids: Collection[str]
) -> bool:
    """Whether okstra opened anything in this pane.

    Any one recorded surface is enough: a pane may hold more than one surface,
    and only one of those UUIDs has to be in the ledger.
    """
    return any(surface_id in owned_surface_ids for surface_id in pane.surface_ids)


def _extend_the_worker_column(workers: Sequence[PaneGeometry]) -> Placement:
    """Split the bottom of the first column downward. A tab would hide the new worker.

    The first worker is split off the lead, so it stays at the lead's y. The
    second and third stack below it. Splitting the bottom parks the new pane at
    the next row rather than inserting it beside the lead.
    """
    bottom = max(workers, key=lambda pane: pane.y)
    return Placement(pane_id=bottom.pane_id, direction="down")


def _extend_the_worker_grid(workers: Sequence[PaneGeometry]) -> Placement:
    """Split worker n-3 to the right, filling the next 3-row column.

    Column-major (x, y) matches creation order while this placement is the only
    one writing the grid: worker 4 is to the right of 1, 5 of 2, 6 of 3, 7 of 4.
    """
    ordered = sorted(workers, key=lambda pane: (pane.x, pane.y))
    source = ordered[len(ordered) - WORKERS_PER_COLUMN]
    return Placement(pane_id=source.pane_id, direction="right")


def plan_worker_height_resize(
    workers: Sequence[PaneGeometry],
) -> WorkerHeightResize | None:
    """The next vertical move that gives every worker an equal share of the column.

    Only the first off-share pane above the bottom is reported. Growing or
    shrinking it changes every pane below, so the caller re-reads and asks
    again. The bottom pane is never the one named: it absorbs the remainder.

    `new-split` halves the pane it lands on and does not take a ratio, so a
    third worker split off the bottom would otherwise stay at a quarter while
    the top one keeps a half.

    A second column inherits row height from the pane it split. Equalizing
    every worker as one stack would move the first column independently of the
    pane sitting in the same row to the right.
    """
    if len(workers) < 2:
        return None
    if len({pane.x for pane in workers}) > 1:
        return None
    ordered = sorted(workers, key=lambda pane: pane.y)
    total = sum(pane.height_points for pane in ordered)
    if total <= 0:
        return None
    share = total / len(ordered)
    for index, pane in enumerate(ordered[:-1]):
        delta = round(pane.height_points - share)
        if delta == 0:
            continue
        if delta < 0:
            return WorkerHeightResize(pane.pane_id, "down", -delta)
        return WorkerHeightResize(ordered[index + 1].pane_id, "up", delta)
    return None


def shim_free_login_path() -> str:
    """The user's login PATH with cmux's per-surface CLI shims removed.

    A cmux pane execs its command through `login … bash --noprofile --norc`, so
    none of the user's shell rc runs and PATH is cmux's own short list. What that
    list does hold is a shim directory bound to the new surface, where `claude`
    and `codex` are wrappers into cmux's agent lifecycle rather than the real
    CLIs — and where a provider cmux does not know has no entry at all, which is
    a plain exit 127 inside the worker wrapper.

    Restoring the login shell's PATH and dropping the shim entries gives the
    wrapper what it would see in an ordinary terminal, which is what its own
    `command -v <cli>` check expects. Nothing here is provider-specific, so
    adding a provider does not touch this path.
    """
    entries = _login_shell_path().split(os.pathsep)
    return os.pathsep.join(
        entry for entry in entries if entry and SHIM_DIR_MARKER not in entry
    )


def worker_command_line(
    *, cwd: Path, argv: Sequence[str], path_value: str
) -> str:
    """The single shell line a cmux pane execs to run one worker.

    Every part is shell-quoted. PATH entries routinely contain spaces — macOS
    ships `/Applications/VMware Fusion.app/Contents/Public` on any machine with
    VMware — and an unquoted assignment stops at the first space, then hands the
    remainder to the shell as a command name.

    The `cd` is part of the command because neither `new-split` nor
    `respawn-pane` accepts a working directory the way `tmux split-window -c`
    does.

    `exec` is here for termination, not for the pane's life, and removing it
    does not keep a finished worker's pane on screen. Measured 2026-08-21: a
    surface's life is bound to the command `respawn-pane` starts — cmux runs it
    through `login … bash --noprofile --norc` and closes the surface when it
    exits — and cmux's settings schema carries no option that keeps a surface
    open past its process. Dropping `exec` was tried against a live dispatch and
    the pane still went with the worker; what it changed was that
    `close_surface` would kill an intermediate shell rather than the worker
    itself. So the adapter contract's "finished workers leave their panes
    behind" does not hold under cmux, and no edit to this line makes it hold.
    """
    return (
        f"cd {shlex.quote(str(cwd))} && "
        f"PATH={shlex.quote(path_value)} exec {shlex.join(argv)}"
    )


def list_panes(workspace: str) -> list[PaneGeometry]:
    return _panes_from(_pane_list_payload(workspace))


def spawn_worker_surface(
    *,
    workspace: str,
    cwd: Path,
    command: Sequence[str],
    title: str,
    owned_surface_ids: Collection[str],
) -> str:
    """Start one worker beside the lead and return its surface UUID.

    The UUID is what okstra records and later closes by. Positional refs cannot
    serve that purpose: cmux renumbers them as surfaces open and close, so a
    close by ref can land on a pane okstra never created.

    `owned_surface_ids` are the UUIDs earlier dispatches returned. They are what
    keeps this placement inside okstra's own panes; the workspace belongs to the
    user and may hold anything.
    """
    panes = list_panes(workspace)
    lead = _lead_pane(panes)
    placement = plan_worker_placement(
        panes,
        lead_pane_id=lead.pane_id,
        owned_surface_ids=owned_surface_ids,
    )
    target = _pane_by_id(panes, placement.pane_id)
    surface_uuid = _open_worker_surface(workspace, placement, target)
    run_cmux(["rename-tab", "--surface", surface_uuid, "--title", title])
    _exec_worker(surface_uuid, cwd=cwd, command=command)
    owned = (*owned_surface_ids, surface_uuid)
    _size_lead_pane(workspace, owned)
    _equalize_worker_heights(workspace, owned)
    return surface_uuid


def close_surface(surface_uuid: str) -> None:
    """Close an okstra-created surface, killing whatever still runs inside it."""
    try:
        run_cmux(["close-surface", "--surface", surface_uuid])
    except (OSError, subprocess.SubprocessError):
        return


def restore_lead_width() -> None:
    """Put the lead back on its target width once its workers are reclaimed.

    Closing a pane hands its width to whichever neighbour cmux picks, and that
    is not necessarily the lead. Measured 2026-08-14: every worker surface of
    the run was gone and the lead was still at 18 columns, because the only
    place that sizes it runs when a worker *opens*. Between rounds — which is
    the stretch the user spends reading the lead rather than the workers — the
    lead therefore kept whatever the last split left it.

    Resolves the workspace itself: teardown reaches this from the CLI, which
    holds a run manifest rather than the workspace UUID that `pane.resize`
    needs. An unresolvable workspace means cmux is gone or was never there, and
    there is no pane left to size.

    Owning nothing is the point of the empty ledger. Teardown has just closed
    every surface it opened, so no pane on screen is okstra's but the lead's —
    which is what makes the lead's share the whole workspace. A pane teardown
    failed to close counts as a stranger's and keeps its width, which is the
    safe way to be wrong here.
    """
    workspace = resolve_lead_workspace()
    if not workspace:
        return
    _size_lead_pane(workspace, ())


def capture_surface(surface_uuid: str, *, last_lines: int = 200) -> str:
    """What the worker's screen shows — for the lead to look at, never to parse.

    cmux hands back a rendered grid: wrapped to the pane's width with a finite
    history, so a path or a count read off it can be silently truncated.
    """
    try:
        result = run_cmux(
            [
                "capture-pane",
                "--surface",
                surface_uuid,
                "--scrollback",
                "--lines",
                str(last_lines),
            ]
        )
    except (OSError, subprocess.SubprocessError):
        return ""
    return result.stdout if result.returncode == 0 else ""


def socket_reachability() -> str:
    """Why this process can or cannot reach cmux, decided at syscall level.

    Separating a sandbox from a closed app without matching on cmux's error
    text: a denied connect raises PermissionError, an absent socket raises
    FileNotFoundError. The distinction decides whether degrading is worth
    attempting at all.
    """
    path = os.environ.get("CMUX_SOCKET_PATH", "").strip()
    if not path:
        return SOCKET_MISSING
    probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        probe.settimeout(SOCKET_PROBE_TIMEOUT_SECONDS)
        probe.connect(path)
        return SOCKET_OK
    except PermissionError:
        return SOCKET_DENIED
    except FileNotFoundError:
        return SOCKET_MISSING
    except OSError:
        return SOCKET_UNREACHABLE
    finally:
        probe.close()


def unreachable_reason() -> str:
    """Why cmux cannot be reached from here, or "" when it can.

    `environment` is the case a socket probe alone cannot see: a sandbox that
    sanitizes the environment leaves no CMUX_* variables at all, so cmux looks
    identical to a machine that never had it — except this run's manifest says
    prepare reached it minutes ago.
    """
    if not _lead_workspace_env():
        return LOST_ENVIRONMENT
    reachability = socket_reachability()
    if reachability == SOCKET_DENIED:
        return LOST_DENIED
    if reachability != SOCKET_OK:
        return LOST_GONE
    return LOST_NOTHING if resolve_lead_workspace() else LOST_GONE


def sidebar_log(workspace: str, message: str, *, level: str = "info") -> None:
    """Append one line to the workspace sidebar's log."""
    _sidebar_call(
        workspace,
        ["log", "--workspace", workspace, "--level", level,
         "--source", SIDEBAR_SOURCE, "--", message],
    )


def sidebar_notify(workspace: str, *, title: str, body: str) -> None:
    """Raise a notification, which badges the tab even from another workspace."""
    _sidebar_call(
        workspace,
        ["notify", "--workspace", workspace, "--title", title, "--body", body],
    )


def sidebar_progress(workspace: str, done: int, total: int, *, label: str) -> None:
    """Show `done`/`total` as the sidebar's progress bar.

    An empty roster reports nothing rather than zero percent: no workers is not
    the same claim as no progress.
    """
    if total <= 0:
        return
    fraction = min(1.0, done / total)
    _sidebar_call(
        workspace,
        ["set-progress", str(round(fraction, 2)), "--workspace", workspace,
         "--label", label],
    )


def _sidebar_call(workspace: str, args: Sequence[str]) -> None:
    """Best-effort sidebar write.

    The sidebar is a view of the run, not part of it, so a wedged or closed cmux
    must never take a dispatch down. The workspace is always explicit: letting
    cmux choose would drop one run's noise into whichever workspace happens to
    be focused, which on a machine running two okstra tasks is the other one.
    """
    if not workspace:
        return
    try:
        run_cmux(args)
    except (OSError, subprocess.SubprocessError):
        return


def rpc(method: str, params: dict[str, Any]) -> dict[str, Any]:
    """Call a cmux rpc method.

    The rpc surface is used rather than the same-named CLI verbs wherever both
    exist, because the CLI swallows failures — `resize-pane` reports success on
    a rejected resize, while `pane.resize` returns the reason.
    """
    result = run_cmux(["rpc", method, json.dumps(params)])
    if result.returncode != 0:
        raise RuntimeError(result.stderr.strip() or f"cmux {method} failed")
    try:
        payload = json.loads(result.stdout)
    except ValueError as exc:
        raise RuntimeError(f"cmux {method} returned non-JSON output") from exc
    return payload if isinstance(payload, dict) else {}


def _open_worker_surface(
    workspace: str, placement: Placement, target: PaneGeometry
) -> str:
    before = open_surface_ids(workspace)
    _create_surface(workspace, placement, target)
    new_ids = open_surface_ids(workspace) - before
    if len(new_ids) != 1:
        raise RuntimeError(
            f"cmux opened {len(new_ids)} surfaces where exactly one was expected"
        )
    return new_ids.pop()


def open_surface_ids(workspace: str) -> set[str]:
    """Every surface UUID the workspace currently holds.

    Dispatch diffs this set across a create to identify the new surface, rather
    than translating the `OK surface:N` echo, because `list-pane-surfaces`
    reports only the focused pane unless given a `--pane`, and the new pane is
    not focused. Teardown intersects its recorded ids with it to tell a surface
    that is still open from one that closed earlier in the run.
    """
    return {
        surface_id
        for pane in list_panes(workspace)
        for surface_id in pane.surface_ids
    }


def _create_surface(
    workspace: str, placement: Placement, target: PaneGeometry
) -> None:
    created = run_cmux(
        [
            "new-split",
            placement.direction,
            "--workspace",
            workspace,
            "--surface",
            target.selected_surface_id or target.surface_ids[0],
        ]
    )
    if created.returncode != 0:
        raise RuntimeError(created.stderr.strip() or "cmux could not open a pane")


def _exec_worker(surface_uuid: str, *, cwd: Path, command: Sequence[str]) -> None:
    line = worker_command_line(
        cwd=cwd, argv=command, path_value=shim_free_login_path()
    )
    started = run_cmux(["respawn-pane", "--surface", surface_uuid, "--command", line])
    if started.returncode != 0:
        raise RuntimeError(started.stderr.strip() or "cmux could not start the worker")


def _size_lead_pane(workspace: str, owned_surface_ids: Collection[str]) -> None:
    """Move the lead's border to its share of the workspace.

    This is the horizontal border okstra places. The first worker column
    inherits whatever the lead leaves. A fourth worker also drops the lead
    from two fifths to two sixths, and that extra sixth goes to the neighbour
    that pulls the border in. Worker heights are equalized separately after
    the split that just halved one of them.

    Which pane carries the request follows from what `pane.resize` does: it
    moves the named pane's own border in the direction given. The lead can push
    its right border out — that is `right` on the lead itself — but it cannot
    pull that border in, because `left` on the leftmost pane finds no adjacent
    border to move. Narrowing is therefore the right-hand neighbour's request,
    and widening is the lead's.

    Run after every worker opens rather than once per round: the split that just
    happened is what knocked the lead off its share, and no other event does.
    """
    payload = _pane_list_payload(workspace)
    container_width = _container_width_points(payload)
    if container_width <= 0:
        # Without the frame the shares are taken of, there is no target to move
        # toward — and a guessed one would move the border to a wrong place
        # rather than leave it where the user last saw it.
        return
    panes = _panes_from(payload)
    lead = _lead_pane(panes)
    offset = lead_resize_points(
        lead,
        target_width_points=lead_target_width(
            panes,
            lead_pane_id=lead.pane_id,
            owned_surface_ids=owned_surface_ids,
            container_width_points=container_width,
        ),
    )
    if offset == 0:
        return
    neighbours = [pane for pane in panes if pane.x > lead.x]
    if not neighbours:
        return
    narrowing = offset > 0
    rpc(
        "pane.resize",
        {
            "workspace_id": workspace,
            "pane_id": (
                min(neighbours, key=lambda pane: pane.x).pane_id
                if narrowing
                else lead.pane_id
            ),
            "direction": "left" if narrowing else "right",
            "amount": abs(offset),
        },
    )


def _equalize_worker_heights(
    workspace: str, owned_surface_ids: Collection[str]
) -> None:
    """Give every owned worker the same height in the first column.

    `new-split` halves whichever pane it lands on, so the third worker of a
    round would otherwise sit at a quarter with the top one still at a half.
    Each call moves one border and re-reads: growing the top pane steals from
    the one below, and that is what the next plan has to see. A second column
    is left alone — its row heights came from the pane it split.
    """
    for _ in range(MAX_WORKER_HEIGHT_MOVES):
        panes = list_panes(workspace)
        lead = _lead_pane(panes)
        workers = [
            pane
            for pane in panes
            if pane.pane_id != lead.pane_id
            and _holds_an_okstra_surface(pane, owned_surface_ids)
        ]
        move = plan_worker_height_resize(workers)
        if move is None:
            return
        rpc(
            "pane.resize",
            {
                "workspace_id": workspace,
                "pane_id": move.pane_id,
                "direction": move.direction,
                "amount": move.amount,
            },
        )


def _lead_pane(panes: Sequence[PaneGeometry]) -> PaneGeometry:
    caller_pane_ref = identify_caller().get("pane_ref", "")
    for pane in panes:
        if caller_pane_ref and pane.ref == caller_pane_ref:
            return pane
    raise RuntimeError("cmux could not locate the lead's pane in its workspace")


def _pane_by_id(panes: Sequence[PaneGeometry], pane_id: str) -> PaneGeometry:
    for pane in panes:
        if pane.pane_id == pane_id:
            return pane
    raise RuntimeError(f"cmux pane {pane_id} disappeared while placing a worker")


def _pane_list_payload(workspace: str) -> dict[str, Any]:
    return rpc("pane.list", {"workspace_id": workspace})


def _panes_from(payload: dict[str, Any]) -> list[PaneGeometry]:
    return [_pane_geometry(entry) for entry in payload.get("panes", [])]


def _container_width_points(payload: dict[str, Any]) -> float:
    """The workspace's own width, which every share here is taken of.

    Reported once per `pane.list` reply rather than per pane, because it is the
    frame the panes are laid out inside — summing the panes would instead give
    whatever they currently happen to occupy.
    """
    frame = payload.get("container_frame") or {}
    return float(frame.get("width") or 0)


def _pane_geometry(entry: dict[str, Any]) -> PaneGeometry:
    frame = entry.get("pixel_frame") or {}
    return PaneGeometry(
        pane_id=str(entry.get("id", "")),
        surface_ids=tuple(str(s) for s in entry.get("surface_ids") or ()),
        columns=int(entry.get("columns", 0)),
        rows=int(entry.get("rows", 0)),
        x=int(frame.get("x", 0)),
        y=int(frame.get("y", 0)),
        cell_width_points=int(entry.get("cell_width_points", 0)),
        width_points=float(frame.get("width") or 0),
        ref=str(entry.get("ref", "")),
        selected_surface_id=str(entry.get("selected_surface_id", "")),
        height_points=float(frame.get("height") or 0),
    )


def _login_shell_path() -> str:
    """PATH as the user's own login shell resolves it.

    Falls back to the inherited PATH: a worker started with a shim-filtered
    inherited PATH is still better than one started with cmux's bare list.
    """
    shell = os.environ.get("SHELL", "") or "/bin/sh"
    try:
        result = subprocess.run(
            [shell, "-lc", 'printf %s "$PATH"'],
            capture_output=True,
            text=True,
            timeout=LOGIN_SHELL_TIMEOUT_SECONDS,
            check=False,
        )
    except (OSError, subprocess.SubprocessError):
        return os.environ.get("PATH", "")
    if result.returncode != 0 or not result.stdout.strip():
        return os.environ.get("PATH", "")
    return result.stdout.strip()


def identify_caller() -> dict[str, Any]:
    """The caller's ref bundle, or {} when cmux cannot resolve it.

    `identify` exits 0 even for an id it cannot resolve, so the exit code says
    nothing; a null `caller` is the only signal that the lookup failed.
    """
    try:
        result = run_cmux(["identify"])
    except (OSError, subprocess.SubprocessError):
        return {}
    if result.returncode != 0:
        return {}
    try:
        payload = json.loads(result.stdout)
    except ValueError:
        return {}
    caller = payload.get("caller")
    return caller if isinstance(caller, dict) else {}


def _ping_answers() -> bool:
    try:
        result = run_cmux(["ping"])
    except (OSError, subprocess.SubprocessError):
        return False
    return result.returncode == 0 and result.stdout.strip() == PING_OK