API reference¶

sessionmux is a terminal application; this reference documents its internal modules for contributors. The entry point is sessionmux.app:main (the sessionmux command).

Sessions¶

Discovering, encoding and tracking Claude Code conversations.

sessionmux.sessions ¶

Data layer for the Claude Code session switcher.

Pure stdlib. Models a session transcript and the sidecar metadata (label, group, tags) we keep apart from Claude's own files, in the XDG data directory (~/.local/share/sessionmux/), with the trash.

Meta `dataclass` ¶

Our own metadata for one session. Stored in the sidecar, never in the JSONL.

Source code in src/sessionmux/sessions.py

@dataclass(frozen=True)
class Meta:
    """Our own metadata for one session. Stored in the sidecar, never in the JSONL."""

    label: str = ""
    group: str = ""
    tags: tuple[str, ...] = ()

Filter `dataclass` ¶

Which conversations the list shows: group is one of groups (when any) AND at least one tag is in tags (when any). Names compare without case.

Source code in src/sessionmux/sessions.py

@dataclass(frozen=True)
class Filter:
    """Which conversations the list shows: group is one of `groups` (when any) AND at least
    one tag is in `tags` (when any). Names compare without case."""

    groups: tuple[str, ...] = ()
    tags: tuple[str, ...] = ()

    @property
    def active(self) -> bool:
        return bool(self.groups or self.tags)

    def matches(self, meta: Meta) -> bool:
        if self.groups and meta.group.casefold() not in {g.casefold() for g in self.groups}:
            return False
        if self.tags:
            want = {t.casefold() for t in self.tags}
            return any(t.casefold() in want for t in meta.tags)
        return True

    def text(self) -> str:
        """As typed in the `f` minibuffer: `g:work t:urgent` (quoted when a name has spaces)."""
        words = [f"g:{g}" for g in self.groups] + [f"t:{t}" for t in self.tags]
        return " ".join(shlex.quote(w) for w in words)

    @classmethod
    def parse(cls, text: str) -> Filter:
        """The inverse of `text`. Raises ValueError on a word without `g:` / `t:`."""
        groups: list[str] = []
        tags: list[str] = []
        for word in shlex.split(text):
            kind, _, name = word.partition(":")
            kind, name = kind.lower(), name.strip()
            if kind not in ("g", "t") or not name:
                raise ValueError(f"not g:<group> or t:<tag>: {word}")
            (groups if kind == "g" else tags).append(name)
        return cls(groups=tuple(groups), tags=tuple(tags))

text ¶

text() -> str

As typed in the f minibuffer: g:work t:urgent (quoted when a name has spaces).

Source code in src/sessionmux/sessions.py

def text(self) -> str:
    """As typed in the `f` minibuffer: `g:work t:urgent` (quoted when a name has spaces)."""
    words = [f"g:{g}" for g in self.groups] + [f"t:{t}" for t in self.tags]
    return " ".join(shlex.quote(w) for w in words)

parse `classmethod` ¶

parse(text: str) -> Filter

The inverse of text. Raises ValueError on a word without g: / t:.

Source code in src/sessionmux/sessions.py

@classmethod
def parse(cls, text: str) -> Filter:
    """The inverse of `text`. Raises ValueError on a word without `g:` / `t:`."""
    groups: list[str] = []
    tags: list[str] = []
    for word in shlex.split(text):
        kind, _, name = word.partition(":")
        kind, name = kind.lower(), name.strip()
        if kind not in ("g", "t") or not name:
            raise ValueError(f"not g:<group> or t:<tag>: {word}")
        (groups if kind == "g" else tags).append(name)
    return cls(groups=tuple(groups), tags=tuple(tags))

Session `dataclass` ¶

One transcript on disk, read-only.

Source code in src/sessionmux/sessions.py

@dataclass(frozen=True)
class Session:
    """One transcript on disk, read-only."""

    session_id: str
    path: Path
    cwd: str  # launch dir (first `user` line)
    last_cwd: str  # latest cwd recorded; follows `cd` during the session
    git_branch: str
    first_prompt: str
    ai_title: str
    custom_title: str
    continued_in: str  # `continued-in` line: ← in Claude moved the conversation to this id
    has_messages: bool  # False for a continuation's stub (title lines only)
    mtime: float
    size: int

Row `dataclass` ¶

One conversation — a continued-in chain, origin first — joined with our metadata.

Pressing ← inside Claude continues the conversation under a new session id (a background session). The chain keeps those ids together as one row.

Source code in src/sessionmux/sessions.py

@dataclass(frozen=True)
class Row:
    """One conversation — a `continued-in` chain, origin first — joined with our metadata.

    Pressing ← inside Claude continues the conversation under a new session id
    (a background session). The chain keeps those ids together as one row.
    """

    chain: tuple[Session, ...]
    meta: Meta
    meta_key: str  # sidecar key: the chain id that already has metadata, else the origin

    @property
    def head(self) -> Session:
        """The newest session of the conversation: the one to open."""
        return self.chain[-1]

    @property
    def ids(self) -> tuple[str, ...]:
        return tuple(s.session_id for s in self.chain)

    def _newest(self, attr: str) -> str:
        return next((getattr(s, attr) for s in reversed(self.chain) if getattr(s, attr)), "")

    @property
    def title(self) -> str:
        """Best human-readable name Claude itself knows."""
        return (
            self._newest("custom_title")
            or self._newest("ai_title")
            or self.chain[0].first_prompt
            or self.head.session_id[:8]
        )

    @property
    def display_title(self) -> str:
        return self.meta.label or self.title

    @property
    def cwd(self) -> str:
        """Where the conversation is working now."""
        return self._newest("last_cwd") or self._newest("cwd")

    @property
    def git_branch(self) -> str:
        return self._newest("git_branch")

    @property
    def mtime(self) -> float:
        return max(s.mtime for s in self.chain)

    @property
    def resumable(self) -> Session:
        """Newest session with real messages: what `claude --resume` can load."""
        return next((s for s in reversed(self.chain) if s.has_messages), self.chain[0])

    @property
    def age(self) -> str:
        secs = max(0.0, time.time() - self.mtime)
        for unit, size in (("d", 86400), ("h", 3600), ("m", 60)):
            if secs >= size:
                return f"{int(secs // size)}{unit}"
        return f"{int(secs)}s"

head `property` ¶

head: Session

The newest session of the conversation: the one to open.

title `property` ¶

title: str

Best human-readable name Claude itself knows.

cwd `property` ¶

cwd: str

Where the conversation is working now.

resumable `property` ¶

resumable: Session

Newest session with real messages: what claude --resume can load.

Read `dataclass` ¶

One transcript as far as it was read. Claude only appends to a transcript, so the next read starts at end and adds the new lines to session.

Source code in src/sessionmux/sessions.py

@dataclass(frozen=True)
class Read:
    """One transcript as far as it was read. Claude only appends to a transcript, so the
    next read starts at `end` and adds the new lines to `session`."""

    stamp: tuple[int, int]  # (mtime_ns, size) of the file when it was read
    end: int  # bytes read: up to the end of the last complete line
    session: Session

Cancelled ¶

Bases: Exception

The scan was cancelled (the app is closing).

Source code in src/sessionmux/sessions.py

class Cancelled(Exception):
    """The scan was cancelled (the app is closing)."""

config_dir ¶

config_dir() -> Path

Root of Claude's config (honours CLAUDE_CONFIG_DIR, like the CLI does).

Source code in src/sessionmux/sessions.py

def config_dir() -> Path:
    """Root of Claude's config (honours CLAUDE_CONFIG_DIR, like the CLI does)."""
    env = os.environ.get("CLAUDE_CONFIG_DIR")
    return Path(env).expanduser() if env else Path.home() / ".claude"

encode_cwd ¶

encode_cwd(cwd: str) -> str

A working dir maps to a project folder: every non-alphanumeric -> '-'.

/home/jcardona/stuff/projects -> -home-jcardona-stuff-projects

Source code in src/sessionmux/sessions.py

def encode_cwd(cwd: str) -> str:
    """A working dir maps to a project folder: every non-alphanumeric -> '-'.

    /home/jcardona/stuff/projects -> -home-jcardona-stuff-projects
    """
    return re.sub(r"[^a-zA-Z0-9]", "-", cwd)

project_dirs ¶

project_dirs(cwd: str) -> list[Path]

The folders of one working directory: Claude names the folder after the real path (symlinks resolved); older transcripts may sit under the path as given.

Source code in src/sessionmux/sessions.py

def project_dirs(cwd: str) -> list[Path]:
    """The folders of one working directory: Claude names the folder after the real path
    (symlinks resolved); older transcripts may sit under the path as given."""
    return list(dict.fromkeys([project_dir(cwd), project_dir(os.path.realpath(cwd))]))

folder_dirs ¶

folder_dirs(folder: str, limit: int = 500) -> list[str]

Existing directories whose path encodes to this folder name, found by walking the filesystem from / (the encoding is lossy: '-' was '/', '.', '_' or '-'). No symlinks on the way, so each one's real path is itself.

Source code in src/sessionmux/sessions.py

def folder_dirs(folder: str, limit: int = 500) -> list[str]:
    """Existing directories whose path encodes to this folder name, found by walking the
    filesystem from / (the encoding is lossy: '-' was '/', '.', '_' or '-'). No symlinks
    on the way, so each one's real path is itself."""
    found: list[str] = []
    todo = [("/", folder[1:])] if folder.startswith("-") else []
    while todo and limit > 0:
        base, rest = todo.pop()
        limit -= 1
        try:
            entries = list(os.scandir(base))
        except OSError:
            continue
        for e in entries:
            if not e.is_dir(follow_symlinks=False):
                continue
            name = encode_cwd(e.name)
            if rest == name:
                found.append(e.path)
            elif rest.startswith(name + "-"):
                todo.append((e.path, rest[len(name) + 1 :]))
    return sorted(found)

launch_dir ¶

launch_dir(s: Session) -> str | None

A directory to start Claude in for this transcript; None when none exists any more.

claude --resume from any other directory still loads the conversation, but Claude then works in that other directory. Claude keeps a transcript in the folder named after the real path of the directory it runs in, so this is a directory whose real path gives the transcript's folder name: the recorded cwd if it does (it need not, for older transcripts), else one found from the folder name.

Source code in src/sessionmux/sessions.py

def launch_dir(s: Session) -> str | None:
    """A directory to start Claude in for this transcript; None when none exists any more.

    `claude --resume` from any other directory still loads the conversation, but Claude
    then works in that other directory. Claude keeps a transcript in the folder named
    after the real path of the directory it runs in, so this is a directory whose real
    path gives the transcript's folder name: the recorded cwd if it does (it need not,
    for older transcripts), else one found from the folder name.
    """
    folder = s.path.parent.name
    for d in (s.cwd, s.last_cwd):
        if d and os.path.isdir(d) and encode_cwd(os.path.realpath(d)) == folder:
            return d
    found = folder_dirs(folder)
    return found[0] if found else None

data_dir ¶

data_dir() -> Path

Our own files: $XDG_DATA_HOME/sessionmux (default ~/.local/share).

Source code in src/sessionmux/sessions.py

def data_dir() -> Path:
    """Our own files: $XDG_DATA_HOME/sessionmux (default ~/.local/share)."""
    return xdgdirs.data_home() / xdgdirs.APP

sidecar_path ¶

sidecar_path() -> Path

Labels, groups and tags, keyed by session id.

Source code in src/sessionmux/sessions.py

def sidecar_path() -> Path:
    """Labels, groups and tags, keyed by session id."""
    return data_dir() / "meta.json"

trash_dir ¶

trash_dir() -> Path

Where deleted conversations go (recoverable: see RESTORE.txt in each folder).

Source code in src/sessionmux/sessions.py

def trash_dir() -> Path:
    """Where deleted conversations go (recoverable: see RESTORE.txt in each folder)."""
    return data_dir() / "trash"

read_transcript ¶

read_transcript(
    path: Path, prev: Read | None = None
) -> Read | None

Read a transcript; with prev, only the lines appended since (all of it if it shrank).

Source code in src/sessionmux/sessions.py

def read_transcript(path: Path, prev: Read | None = None) -> Read | None:
    """Read a transcript; with `prev`, only the lines appended since (all of it if it shrank)."""
    try:
        st = path.stat()
    except OSError:
        return None
    if prev is not None and st.st_size < prev.end:
        prev = None  # rewritten: read it again from the start
    old = prev.session if prev else None
    session_id = old.session_id if old else ""
    cwd = old.cwd if old else ""
    git_branch = old.git_branch if old else ""
    first_prompt = old.first_prompt if old else ""
    ai_title = old.ai_title if old else ""
    custom_title = old.custom_title if old else ""
    continued_in = old.continued_in if old else ""
    has_messages = old.has_messages if old else False
    end = prev.end if prev else 0

    try:
        with path.open("rb") as fh:
            fh.seek(end)
            for line in fh:
                # a last line without its newline is still being written: it is read
                # now and again next time (every field below takes it twice the same way)
                if line.endswith(b"\n"):
                    end += len(line)
                # Transcripts reach hundreds of MB: look at the start of the raw line and
                # parse JSON only for the few lines that carry what the list shows. In user
                # and title lines Claude writes the top-level "type" before any nested
                # object, within the first 200 bytes (checked on every local transcript:
                # 2.1.302). An assistant line mostly starts with its nested "message", so
                # its first "type" is "message": a conversation counts as having messages
                # by its user lines (and the few assistant lines written type-first).
                if not session_id and (m := _SESSION_ID.search(line)):
                    session_id = m.group(1).decode()
                m = _TYPE.search(line, 0, 200)
                kind = m.group(1) if m else b""
                if kind in (b"user", b"assistant"):
                    has_messages = True
                wanted = (
                    kind == b"user" and not (cwd and git_branch and first_prompt)
                ) or kind in _PARSED_TYPES
                if not wanted:
                    continue
                try:
                    d = json.loads(line)
                except json.JSONDecodeError:
                    continue
                t = d.get("type")
                if t == "ai-title" and d.get("aiTitle"):
                    ai_title = d["aiTitle"]
                elif t == "custom-title" and d.get("customTitle"):
                    custom_title = d["customTitle"]
                elif t == "continued-in" and d.get("continuedInSessionId"):
                    continued_in = d["continuedInSessionId"]
                elif t == "user":
                    if not cwd and d.get("cwd"):
                        cwd = d["cwd"]
                    if not git_branch and d.get("gitBranch"):
                        git_branch = d["gitBranch"]
                    if not first_prompt:
                        msg = d.get("message") or {}
                        txt = _text_of(msg.get("content")) if isinstance(msg, dict) else ""
                        # skip tool-result / slash-command noise for the preview
                        if txt and not txt.startswith(("<", "/")):
                            first_prompt = txt.strip()
        last_cwd = _last_cwd(path)
    except OSError:
        return None

    session = Session(
        session_id=session_id or path.stem,
        path=path,
        cwd=cwd,
        last_cwd=last_cwd,
        git_branch=git_branch,
        first_prompt=first_prompt,
        ai_title=ai_title,
        custom_title=custom_title,
        continued_in=continued_in,
        has_messages=has_messages,
        mtime=st.st_mtime,
        size=st.st_size,
    )
    return Read(stamp=(st.st_mtime_ns, st.st_size), end=end, session=session)

scan ¶

scan(
    cwd: str | None, cancelled: Callable[[], bool] = _never
) -> list[Session]

All transcripts of one working directory (None: of every directory), newest first.

A transcript whose mtime and size did not change is not read again; one that grew is read from where the last read stopped. cancelled is checked before each file (raises Cancelled), so a first read of every directory does not hold up a quit.

Source code in src/sessionmux/sessions.py

def scan(cwd: str | None, cancelled: Callable[[], bool] = _never) -> list[Session]:
    """All transcripts of one working directory (None: of every directory), newest first.

    A transcript whose mtime and size did not change is not read again; one that grew
    is read from where the last read stopped. `cancelled` is checked before each file
    (raises Cancelled), so a first read of every directory does not hold up a quit.
    """
    with _scan_lock:
        return _scan(cwd, cancelled)

load_meta ¶

load_meta() -> dict[str, Meta]

The sidecar; entries of the wrong shape are skipped (it is a file people can edit).

Source code in src/sessionmux/sessions.py

def load_meta() -> dict[str, Meta]:
    """The sidecar; entries of the wrong shape are skipped (it is a file people can edit)."""
    return read_meta(sidecar_path())

save_meta ¶

save_meta(meta: dict[str, Meta]) -> None

Write the sidecar atomically: a crash leaves the old file or the new one, never half.

Source code in src/sessionmux/sessions.py

def save_meta(meta: dict[str, Meta]) -> None:
    """Write the sidecar atomically: a crash leaves the old file or the new one, never half."""
    path = sidecar_path()
    xdgdirs.make_dirs(path.parent, xdgdirs.data_home())
    payload = {
        sid: {"label": m.label, "group": m.group, "tags": list(m.tags)}
        for sid, m in meta.items()
        if m.label or m.group or m.tags
    }
    tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp")
    tmp.write_text(json.dumps(payload, indent=2), encoding="utf-8")
    os.replace(tmp, path)

meta_locked ¶

meta_locked() -> Generator[None]

Hold the sidecar for a read-modify-write: excludes other threads and other app instances (flock on a lock file next to it).

Source code in src/sessionmux/sessions.py

@contextmanager
def meta_locked() -> Generator[None]:
    """Hold the sidecar for a read-modify-write: excludes other threads and other app
    instances (flock on a lock file next to it)."""
    lock = sidecar_path().with_name(sidecar_path().name + ".lock")
    xdgdirs.make_dirs(lock.parent, xdgdirs.data_home())
    with lock.open("a") as fh:
        fcntl.flock(fh, fcntl.LOCK_EX)
        yield  # closing the file releases the lock

update_meta ¶

update_meta(
    key: str, change: Callable[[Meta], Meta]
) -> None

Change one session's metadata.

Source code in src/sessionmux/sessions.py

def update_meta(key: str, change: Callable[[Meta], Meta]) -> None:
    """Change one session's metadata."""
    with meta_locked():
        meta = load_meta()
        meta[key] = change(meta.get(key, Meta()))
        save_meta(meta)

chains ¶

chains(
    sessions: list[Session],
) -> list[tuple[Session, ...]]

Group transcripts into conversations by following continued-in links.

Source code in src/sessionmux/sessions.py

def chains(sessions: list[Session]) -> list[tuple[Session, ...]]:
    """Group transcripts into conversations by following `continued-in` links."""
    by_id = {s.session_id: s for s in sessions}
    targets = {s.continued_in for s in sessions if s.continued_in in by_id}
    origins = [s for s in sessions if s.session_id not in targets]
    placed: set[str] = set()
    out: list[tuple[Session, ...]] = []
    # origins first; then whatever is left (only a broken, cyclic chain) so nothing disappears
    for start in origins + [s for s in sessions if s.session_id in targets]:
        if start.session_id in placed:
            continue
        chain, cur = [start], start
        placed.add(start.session_id)
        while cur.continued_in in by_id and cur.continued_in not in placed:
            cur = by_id[cur.continued_in]
            chain.append(cur)
            placed.add(cur.session_id)
        out.append(tuple(chain))
    return out

build_rows ¶

build_rows(
    cwd: str | None,
    cancelled: Callable[[], bool] = _never,
    only: Filter = NO_FILTER,
) -> list[Row]

The conversations of one working directory (None: of every directory) that only matches.

Source code in src/sessionmux/sessions.py

def build_rows(
    cwd: str | None, cancelled: Callable[[], bool] = _never, only: Filter = NO_FILTER
) -> list[Row]:
    """The conversations of one working directory (None: of every directory) that `only`
    matches."""
    meta = load_meta()
    rows: list[Row] = []
    for chain in chains(scan(cwd, cancelled)):
        key = next((s.session_id for s in chain if s.session_id in meta), chain[0].session_id)
        row_meta = meta.get(key, Meta())
        if only.matches(row_meta):
            rows.append(Row(chain=chain, meta=row_meta, meta_key=key))
    # group together, ungrouped last, then newest-first inside each group
    rows.sort(key=lambda r: (r.meta.group == "", r.meta.group, -r.mtime))
    return rows

trash ¶

trash(row: Row) -> Path

Move the conversation's transcripts to the trash and drop its metadata.

After this, neither the list nor Claude's /resume shows it. Returns the trash folder; its RESTORE.txt says where each file came from.

Source code in src/sessionmux/sessions.py

def trash(row: Row) -> Path:
    """Move the conversation's transcripts to the trash and drop its metadata.

    After this, neither the list nor Claude's `/resume` shows it. Returns the
    trash folder; its RESTORE.txt says where each file came from.
    """
    dest = trash_dir() / f"{time.strftime('%Y%m%d-%H%M%S')}-{row.chain[0].session_id[:8]}"
    xdgdirs.make_dirs(dest, xdgdirs.data_home())
    moved = []
    for s in row.chain:
        if s.path.exists():
            target = dest / s.path.name
            shutil.move(s.path, target)
            moved.append(f"{target} -> {s.path}")
    (dest / "RESTORE.txt").write_text(
        f"{row.display_title}\nMove each file back to restore the conversation:\n"
        + "\n".join(moved)
        + "\n",
        encoding="utf-8",
    )
    drop_meta(row.ids)
    return dest

State¶

On-disk state: labels, the trash, and what stays open across runs.

sessionmux.state ¶

What the app had open, kept so the next start (or the start after a crash) has it again.

Pure stdlib. One small JSON file per list directory, under the XDG state directory ($XDG_STATE_HOME, default ~/.local/state: "current state of the application that can be reused on a restart"). Written on every change, atomically, so it is current even when the computer dies.

OpenState `dataclass` ¶

The conversations open in the app for one list directory.

Source code in src/sessionmux/state.py

@dataclass(frozen=True)
class OpenState:
    """The conversations open in the app for one list directory."""

    cwd: str  # the list directory (the app's own)
    all_dirs: bool = False  # the list showed every directory (`a`)
    filter: Filter = NO_FILTER  # the list showed only these groups / tags (`f`, -G, -t)
    open: tuple[str, ...] = ()  # session ids of the open conversations, in opening order
    shown: str | None = None  # the one in the main pane

load ¶

load(cwd: str) -> OpenState

The saved state of this list directory; an empty one when there is none or it cannot be read (it is only a convenience).

Source code in src/sessionmux/state.py

def load(cwd: str) -> OpenState:
    """The saved state of this list directory; an empty one when there is none or it
    cannot be read (it is only a convenience)."""
    empty = OpenState(cwd=cwd)
    try:
        raw = json.loads(state_path(cwd).read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return empty
    if not isinstance(raw, dict) or raw.get("cwd") != cwd:  # another dir, same encoded name
        return empty
    ids = raw.get("open")
    shown = raw.get("shown")
    return OpenState(
        cwd=cwd,
        all_dirs=raw.get("all_dirs") is True,
        filter=_read_filter(raw.get("filter")),
        open=tuple(i for i in ids if isinstance(i, str) and i) if isinstance(ids, list) else (),
        shown=shown if isinstance(shown, str) and shown else None,
    )

save ¶

save(state: OpenState) -> None

Write atomically (temp file + rename); missing directories are made 0700. Raises OSError.

Source code in src/sessionmux/state.py

def save(state: OpenState) -> None:
    """Write atomically (temp file + rename); missing directories are made 0700.
    Raises OSError."""
    path = state_path(state.cwd)
    make_dirs(path.parent, state_home())
    payload = {
        "cwd": state.cwd,
        "all_dirs": state.all_dirs,
        "filter": {"groups": list(state.filter.groups), "tags": list(state.filter.tags)},
        "open": list(state.open),
        "shown": state.shown,
    }
    tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp")
    tmp.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
    os.replace(tmp, path)

Usage¶

CPU and memory accounting for the running Claude processes.

sessionmux.usage ¶

CPU, memory and wakeups of process trees, from Linux /proc (no root needed).

A Claude session is a tree of processes: claude, its MCP servers and its tool runs. Per tree, between two samples:

CPU: % of one core. Each process's utime + stime, plus what processes that ended in between used: a parent's cutime + cstime grows by the whole life of a child it waited for, minus what the last sample had already counted for that child. So a short tool run the walk never saw still counts.
Memory: anonymous resident memory (statm: resident - shared). It leaves out file pages shared with other processes (libraries), so a sum over a tree is not inflated the way RSS is. On this machine it was within 6% of PSS (smaps_rollup) for 232 Claude processes, at 10 ms instead of 435 ms.
Wakeups per second: how often its threads were put on a CPU (third field of /proc//task//schedstat); waking a CPU from idle is what costs power, the kind of number powertop shows. It reads every thread (~2,400 files for 23 Claudes), so only when asked (the top view).

Finding the tree: a child belongs to the thread that made it. Reading every thread's children file finds all (71 ms for 23 Claudes); the main thread's alone misses a few (8 of 232) at 6 ms. So a round walks every thread only every FULL_WALK_ROUNDS rounds, and in between keeps the processes found before.

Pure stdlib. Nothing here is available off Linux: available() says so.

Proc `dataclass` ¶

One process at one moment.

Source code in src/sessionmux/usage.py

@dataclass(frozen=True)
class Proc:
    """One process at one moment."""

    pid: int
    ppid: int
    comm: str
    start: int  # starttime, ticks since boot: tells a reused pid apart
    ticks: int  # utime + stime, all its threads
    child_ticks: int  # cutime + cstime: its ended children it waited for
    anon_kb: int  # anonymous resident memory
    slices: int | None  # times its threads were put on a CPU; None: not read

    @property
    def key(self) -> tuple[int, int]:
        return self.pid, self.start

Sample `dataclass` ¶

A process tree at one moment.

Source code in src/sessionmux/usage.py

@dataclass(frozen=True)
class Sample:
    """A process tree at one moment."""

    root: int
    at: float  # time.monotonic()
    procs: tuple[Proc, ...]  # the root first

Use `dataclass` ¶

One tree's use between two samples (or several trees together).

Source code in src/sessionmux/usage.py

@dataclass(frozen=True)
class Use:
    """One tree's use between two samples (or several trees together)."""

    cpu: float  # % of one core, ended processes included
    mem_mb: float  # now
    wakeups: float | None  # per second; None: not measured
    procs: tuple[ProcUse, ...]  # busiest first
    ended_cpu: float  # the part of `cpu` used by processes that ended in between

    @staticmethod
    def total(uses: list[Use]) -> Use:
        known = [u.wakeups for u in uses if u.wakeups is not None]
        return Use(
            cpu=sum(u.cpu for u in uses),
            mem_mb=sum(u.mem_mb for u in uses),
            wakeups=sum(known) if known else None,
            procs=tuple(p for u in uses for p in u.procs),
            ended_cpu=sum(u.ended_cpu for u in uses),
        )

Round `dataclass` ¶

One sampling round over every tree: what the next round starts from.

Source code in src/sessionmux/usage.py

@dataclass(frozen=True)
class Round:
    """One sampling round over every tree: what the next round starts from."""

    samples: dict[int, Sample]  # root pid -> its tree now
    uses: dict[int, Use]  # root pid -> its use since the last round
    number: int = 0  # rounds so far (every FULL_WALK_ROUNDS-th walks fully)

children ¶

children(pid: int, every_thread: bool = True) -> list[int]

Its child processes: through every thread's children file, or the main thread's alone (cheaper; misses children another thread made).

Source code in src/sessionmux/usage.py

def children(pid: int, every_thread: bool = True) -> list[int]:
    """Its child processes: through every thread's `children` file, or the main
    thread's alone (cheaper; misses children another thread made)."""
    tids = _threads(pid) if every_thread else [str(pid)]
    out: list[int] = []
    for tid in tids:
        text = _read(f"/proc/{pid}/task/{tid}/children")
        if text:
            out += [int(c) for c in text.split()]
    return out

sample ¶

sample(
    root: int,
    with_children: bool = True,
    *,
    wakeups: bool = False,
    full: bool = True,
    known: Mapping[int, int] | None = None,
) -> Sample | None

The tree under root now; None when root is gone.

full: find children through every thread; else through the main thread, then add the known processes (pid -> start, from the last sample) the walk missed that are still there, under a process of the tree.

Source code in src/sessionmux/usage.py

def sample(
    root: int,
    with_children: bool = True,
    *,
    wakeups: bool = False,
    full: bool = True,
    known: Mapping[int, int] | None = None,
) -> Sample | None:
    """The tree under `root` now; None when `root` is gone.

    full: find children through every thread; else through the main thread, then add
    the `known` processes (pid -> start, from the last sample) the walk missed that are
    still there, under a process of the tree.
    """
    at = time.monotonic()
    first = read_proc(root, wakeups)
    if first is None:
        return None
    procs = [first]
    seen = {root}
    todo = children(root, every_thread=full) if with_children else []

    def walk() -> None:
        while todo:
            pid = todo.pop(0)
            if pid in seen:
                continue
            seen.add(pid)
            proc = read_proc(pid, wakeups)
            if proc is not None:
                procs.append(proc)
                todo.extend(children(pid, every_thread=full))

    walk()
    if with_children and known and not full:
        missed = sorted(pid for pid in known if pid not in seen)  # parents first, mostly
        for pid in missed:
            proc = read_proc(pid, wakeups)
            if proc is not None and proc.start == known[pid] and proc.ppid in seen:
                seen.add(pid)
                procs.append(proc)
                todo.extend(children(pid, every_thread=False))
                walk()
    return Sample(root=root, at=at, procs=tuple(procs))

use ¶

use(prev: Sample, now: Sample) -> Use

What the tree used between prev and now.

Source code in src/sessionmux/usage.py

def use(prev: Sample, now: Sample) -> Use:
    """What the tree used between `prev` and `now`."""
    dt = max(now.at - prev.at, 1e-6)
    before = {p.key: p for p in prev.procs}
    keys = {p.key for p in now.procs}
    procs: list[ProcUse] = []
    reaped = 0
    for p in now.procs:
        q = before.get(p.key)
        ticks = p.ticks - q.ticks if q else p.ticks  # new: started in between
        wake = None
        if p.slices is not None and (q is None or q.slices is not None):
            wake = max(p.slices - (q.slices if q and q.slices is not None else 0), 0) / dt
        if q:
            reaped += p.child_ticks - q.child_ticks
        procs.append(
            ProcUse(
                pid=p.pid,
                comm=p.comm,
                cpu=max(ticks, 0) * 100 / TICKS / dt,
                mem_mb=p.anon_kb / 1024,
                wakeups=wake,
            )
        )
    counted = sum(q.ticks for k, q in before.items() if k not in keys)
    ended = max(reaped - counted, 0) * 100 / TICKS / dt
    procs.sort(key=lambda u: u.cpu, reverse=True)
    known = [u.wakeups for u in procs if u.wakeups is not None]
    return Use(
        cpu=sum(u.cpu for u in procs) + ended,
        mem_mb=sum(u.mem_mb for u in procs),
        wakeups=sum(known) if known else None,
        procs=tuple(procs),
        ended_cpu=ended,
    )

sample_all ¶

sample_all(
    roots: Mapping[int, bool],
    last: Round,
    *,
    wakeups: bool = False,
) -> Round

Sample every tree (root pid -> with its children). A use needs the same root process in the last round (not a reused pid).

Source code in src/sessionmux/usage.py

def sample_all(roots: Mapping[int, bool], last: Round, *, wakeups: bool = False) -> Round:
    """Sample every tree (root pid -> with its children). A use needs the same root
    process in the last round (not a reused pid)."""
    full = last.number % FULL_WALK_ROUNDS == 0
    samples: dict[int, Sample] = {}
    uses: dict[int, Use] = {}
    for root, with_children in roots.items():
        q = last.samples.get(root)
        known = {p.pid: p.start for p in q.procs} if q else None
        s = sample(root, with_children, wakeups=wakeups, full=full, known=known)
        if s is None:
            continue
        samples[root] = s
        if q is not None and q.procs[0].key == s.procs[0].key:
            uses[root] = use(q, s)
    return Round(samples=samples, uses=uses, number=last.number + 1)

Agents¶

Resolving the agent/process tree behind each session.

sessionmux.agents ¶

Live session state from Claude Code's own claude agents --json.

Read-only. Used to mark rows (running / waiting / working), to open background sessions with claude attach, and to tell which session the main pane shows after ← in Claude (the pane title is the shown agent's name).

live_agents ¶

live_agents() -> dict[str, AgentState] | None

session_id -> AgentState; None when claude agents failed (then nothing is known, which is not the same as nothing running). Takes ~0.35 s — call it off the UI thread.

Source code in src/sessionmux/agents.py

def live_agents() -> dict[str, AgentState] | None:
    """session_id -> AgentState; None when `claude agents` failed (then nothing is known,
    which is not the same as nothing running). Takes ~0.35 s — call it off the UI thread."""
    try:
        r = _run("agents", "--json", "--all", timeout=6)
        data = json.loads(r.stdout)
    except (OSError, ValueError, subprocess.SubprocessError):
        return None
    if r.returncode != 0 or not isinstance(data, list):
        return None
    out: dict[str, AgentState] = {}
    for e in data:
        if not isinstance(e, dict):
            continue
        sid = _text(e, "sessionId")
        if sid:
            pid = e.get("pid")
            out[sid] = AgentState(
                session_id=sid,
                agent_id=_text(e, "id") or sid[:8],
                kind=_text(e, "kind"),
                name=_text(e, "name"),
                cwd=_text(e, "cwd"),
                status=_text(e, "status"),
                state=_text(e, "state"),
                pid=pid if isinstance(pid, int) and not isinstance(pid, bool) else None,
            )
    return out

remove ¶

remove(agent_id: str) -> str | None

Stop and delete a background session (its agent-view row, and a worktree Claude made for it when that is safe). None on success, else Claude's reason for refusing, e.g. a worktree with unpushed commits.

Source code in src/sessionmux/agents.py

def remove(agent_id: str) -> str | None:
    """Stop and delete a background session (its agent-view row, and a worktree
    Claude made for it when that is safe). None on success, else Claude's reason
    for refusing, e.g. a worktree with unpushed commits."""
    try:
        _run("stop", agent_id, timeout=20)
        r = _run("rm", agent_id, timeout=20)
    except (OSError, subprocess.SubprocessError) as exc:
        return str(exc)
    if r.returncode == 0:
        return None
    lines = (r.stderr or r.stdout).strip().splitlines()
    return lines[0] if lines else f"`claude rm {agent_id}` failed"

wait_exit ¶

wait_exit(pids: list[int], timeout: float = 5.0) -> bool

Wait until these processes are gone (so none writes its transcript again).

Source code in src/sessionmux/agents.py

def wait_exit(pids: list[int], timeout: float = 5.0) -> bool:
    """Wait until these processes are gone (so none writes its transcript again)."""
    end = time.monotonic() + timeout
    while True:
        alive = [p for p in pids if _alive(p)]
        if not alive:
            return True
        if time.monotonic() > end:
            return False
        time.sleep(0.05)

title_name ¶

title_name(title: str) -> str

The name in a terminal title, without Claude's spinner glyph.

Source code in src/sessionmux/agents.py

def title_name(title: str) -> str:
    """The name in a terminal title, without Claude's spinner glyph."""
    return _SPINNER.sub("", title).strip()

session_for_title ¶

session_for_title(
    title: str,
    agents: dict[str, AgentState],
    current: str | None,
) -> str | None

The session a terminal title names, or None when it names none or several.

Claude sets the title to the shown session's name (with a spinner glyph while working), and to "... · claude agents" in agent view. This is observed behaviour on 2.1.302, not a documented interface.

Source code in src/sessionmux/agents.py

def session_for_title(title: str, agents: dict[str, AgentState], current: str | None) -> str | None:
    """The session a terminal title names, or None when it names none or several.

    Claude sets the title to the shown session's name (with a spinner glyph
    while working), and to "... · claude agents" in agent view. This is
    observed behaviour on 2.1.302, not a documented interface.
    """
    name = title_name(title)
    if not name:
        return None
    if name.endswith("…"):
        stem = name[:-1].rstrip()
        hits = [a.session_id for a in agents.values() if a.name and a.name.startswith(stem)]
    else:
        hits = [a.session_id for a in agents.values() if a.name == name]
    if current in hits:
        return current
    return hits[0] if len(hits) == 1 else None

XDG directories¶

Where sessionmux keeps its state and config.

sessionmux.xdgdirs ¶

The XDG base directories this app writes to (specifications.freedesktop.org/basedir).

data ($XDG_DATA_HOME, default ~/.local/share): labels, groups, tags; the trash.
state ($XDG_STATE_HOME, default ~/.local/state): the open conversations, for the next start.

A value that is not an absolute path is invalid and ignored (the spec). Missing directories are made with mode 0700; existing ones keep their mode.

make_dirs ¶

make_dirs(path: Path, base: Path) -> None

Make path and its missing parents up to and including base, each with mode 0700.

Source code in src/sessionmux/xdgdirs.py

def make_dirs(path: Path, base: Path) -> None:
    """Make `path` and its missing parents up to and including `base`, each with mode 0700."""
    todo = []
    d = path
    while not d.is_dir():
        todo.append(d)
        if d == base or d.parent == d:
            break
        d = d.parent
    if base in todo and not base.parent.is_dir():
        base.parent.mkdir(parents=True, exist_ok=True)  # above the base: not ours
    for d in reversed(todo):
        d.mkdir(mode=0o700, exist_ok=True)

API reference¶

Sessions¶

sessionmux.sessions ¶

Meta dataclass ¶

Filter dataclass ¶

text ¶

parse classmethod ¶

Session dataclass ¶

Row dataclass ¶

head property ¶

title property ¶

cwd property ¶

resumable property ¶

Read dataclass ¶

Cancelled ¶

config_dir ¶

encode_cwd ¶

project_dirs ¶

folder_dirs ¶

launch_dir ¶

data_dir ¶

sidecar_path ¶

trash_dir ¶

read_transcript ¶

scan ¶

load_meta ¶

save_meta ¶

meta_locked ¶

update_meta ¶

chains ¶

build_rows ¶

trash ¶

State¶

sessionmux.state ¶

OpenState dataclass ¶

load ¶

save ¶

Usage¶

sessionmux.usage ¶

Proc dataclass ¶

Sample dataclass ¶

Use dataclass ¶

Round dataclass ¶

children ¶

sample ¶

use ¶

sample_all ¶

Agents¶

sessionmux.agents ¶

live_agents ¶

remove ¶

wait_exit ¶

title_name ¶

session_for_title ¶

XDG directories¶

sessionmux.xdgdirs ¶

make_dirs ¶

Meta `dataclass` ¶

Filter `dataclass` ¶

parse `classmethod` ¶

Session `dataclass` ¶

Row `dataclass` ¶

head `property` ¶

title `property` ¶

cwd `property` ¶

resumable `property` ¶

Read `dataclass` ¶

OpenState `dataclass` ¶

Proc `dataclass` ¶

Sample `dataclass` ¶

Use `dataclass` ¶

Round `dataclass` ¶