feat: reset tmux window indexes

This commit is contained in:
2026-06-10 21:56:21 -04:00
parent 666df40aca
commit bebdf8c1ee
+106
View File
@@ -116,6 +116,15 @@ class WindowInfo:
last_activity: int
@dataclass(frozen=True)
class WindowIndex:
"""Stable tmux window identity plus its current numeric index."""
window_id: str
index: int
name: str
class TmuxRunner:
"""Low-level async tmux command wrapper. All operations target windows within a single session."""
@@ -231,6 +240,103 @@ class TmuxRunner:
flags.extend(["-e", f"{key}={value}"])
return flags
async def _list_window_indexes(self) -> list[WindowIndex] | None:
"""Return window ids, indexes, and names for this tmux session.
None means tmux failed or returned malformed data; an empty list means
the session listed successfully but has no windows.
"""
rc, stdout, err = await self._exec(
"list-windows",
"-t",
self.session_name,
"-F",
"#{window_id}\t#{window_index}\t#{window_name}",
)
if rc != 0:
log.error("list-windows failed for %s: %s", self.session_name, err)
return None
windows: list[WindowIndex] = []
for line in stdout.splitlines():
if not line.strip():
continue
parts = line.split("\t", 2)
if len(parts) != 3:
log.error("Malformed list-windows row: %r", line)
return None
window_id, index_text, name = parts
try:
index = int(index_text)
except ValueError:
log.error("Malformed window index in row: %r", line)
return None
windows.append(WindowIndex(window_id=window_id, index=index, name=name))
return windows
async def _move_window_id_to_index(self, window_id: str, index: int) -> bool:
rc, _, err = await self._exec(
"move-window",
"-s",
window_id,
"-t",
f"{self.session_name}:{index}",
)
if rc != 0:
log.error("move-window failed for %s -> %s: %s", window_id, index, err)
return False
return True
async def reset_window_indexes(self, home_window: str) -> bool:
"""Compact tmux window indexes while keeping the home window at 0.
The operation changes tmux indexes only. It preserves window names,
panes, processes, hqt database rows, and harness session ids.
"""
windows = await self._list_window_indexes()
if windows is None:
return False
home = next((window for window in windows if window.name == home_window), None)
if home is None:
log.error(
"Home window %r not found in session %s",
home_window,
self.session_name,
)
return False
non_home = sorted(
(window for window in windows if window.window_id != home.window_id),
key=lambda window: window.index,
)
desired: list[tuple[WindowIndex, int]] = [(home, 0)]
desired.extend(
(window, index) for index, window in enumerate(non_home, start=1)
)
if all(window.index == target_index for window, target_index in desired):
return True
# Move every window out of the target range before placing final indexes.
# This avoids move-window failures when a destination index is occupied.
temp_start = max(window.index for window in windows) + len(windows)
for offset, window in enumerate(
sorted(windows, key=lambda item: item.index),
start=1,
):
if not await self._move_window_id_to_index(
window.window_id,
temp_start + offset,
):
return False
for window, target_index in desired:
if not await self._move_window_id_to_index(window.window_id, target_index):
return False
return True
async def new_window(
self,
window_name: str,