stem-launcher/rvctl.py

#!/usr/bin/env python3
"""STEM workspace launcher (``stemctl`` with ``rvctl`` compatibility)."""

from __future__ import annotations

import argparse
import base64
import datetime as dt
import hashlib
import json
import os
import re
import shlex
import shutil
import stat
import subprocess
import sys
import time
from dataclasses import dataclass, replace
from pathlib import Path
from urllib import error as urlerror
from urllib import request as urlrequest
from urllib.parse import quote, unquote, urlsplit, urlunsplit


def expand_path(raw_path: str, base_dir: Path) -> Path:
    expanded = os.path.expandvars(os.path.expanduser(raw_path))
    path = Path(expanded)
    if not path.is_absolute():
        path = base_dir / path
    return path.resolve()


@dataclass
class GitConfig:
    base_url: str
    source_org: str
    answer_org: str
    source_remote: str
    answer_remote: str
    origin_remote: str
    fallback_branch: str


@dataclass
class WorkspaceConfig:
    config_path: Path
    original_root: Path
    original_series_root: Path
    workspace_root: Path
    workspace_info_path: Path
    workspace_info_url: str
    workspace_info: dict | None
    series_root: Path
    socket_root: Path
    token_path: Path
    tools_root_candidates: list[Path]
    tools_root: Path
    env_tool_url: str
    defaults: dict[str, str]
    git: GitConfig


@dataclass(frozen=True)
class TaskEntry:
    name: str
    path: Path


@dataclass(frozen=True)
class CardInfo:
    series_id: str
    card_id: str
    repo: str
    branch: str
    title: str
    source_org: str
    answer_org: str
    source_remote: str
    answer_remote: str
    fallback_branch: str
    source_path: Path | None = None


@dataclass(frozen=True)
class SessionTarget:
    series: str
    card: str
    card_path: Path
    task: str
    profile: str
    target: str
    instance: str


SCOPE_FIELDS = [
    ("a", "activitypub"),
    ("A", "admin"),
    ("i", "issue"),
    ("m", "misc"),
    ("n", "notification"),
    ("o", "organization"),
    ("p", "package"),
    ("r", "repository"),
    ("u", "user"),
]

SCOPE_CATEGORY_ALIASES = {
    "organization": ["organization", "org"],
    "repository": ["repository", "repo"],
}

ORG_PERMISSION_FIELDS = [
    ("o", "owner"),
    ("a", "admin"),
    ("w", "write"),
    ("r", "read"),
    ("c", "create"),
]

REPO_PERMISSION_FIELDS = [
    ("o", "owner"),
    ("a", "admin"),
    ("w", "write"),
    ("r", "read"),
]

ENV_PROFILES = {
    "native-amd64": {
        "service": "native-amd64",
        "image": "localhost/stem/dev-native-amd64:local",
        "purpose": "native AMD64 C/C++ tests, sanitizers and GDB/LLDB",
    },
    "hazard3-sim": {
        "service": "hazard3-sim",
        "image": "localhost/stem/dev-hazard3-sim:local",
        "purpose": "Hazard3 RTL, RV32 bare metal/FreeRTOS and GDB",
    },
    "rp2350": {
        "service": "rp2350",
        "image": "localhost/stem/dev-rp2350:local",
        "purpose": "physical RP2350 RISC-V or ARM build, deploy and debug",
    },
}

ENV_PROFILE_ALIASES = {
    "host": "native-amd64",
    "native": "native-amd64",
    "amd64": "native-amd64",
    "rv32i": "hazard3-sim",
    "riscv": "hazard3-sim",
    "riscv32": "hazard3-sim",
    "hazard3": "hazard3-sim",
}

ENV_PROFILE_CHOICES = sorted(set(ENV_PROFILES) | set(ENV_PROFILE_ALIASES))
PROFILE_DEFAULT_TARGETS = {
    "native-amd64": "native",
    "hazard3-sim": "hazard3-baremetal",
    "rp2350": "rp2350-rv",
}
PROFILE_TARGETS = {
    "native-amd64": frozenset({"native"}),
    "hazard3-sim": frozenset({"hazard3-baremetal", "hazard3-freertos"}),
    "rp2350": frozenset({"rp2350-rv", "rp2350-arm"}),
}


def default_runtime_socket_root(workspace_root: Path) -> Path:
    xdg_runtime = os.environ.get("XDG_RUNTIME_DIR", "").strip()
    if xdg_runtime:
        return Path(xdg_runtime).expanduser().resolve() / "stem"
    run_user = Path(f"/run/user/{os.getuid()}")
    if run_user.is_dir():
        return run_user / "stem"
    return workspace_root / ".runtime" / "stem"


def load_config(config_path: Path) -> WorkspaceConfig:
    raw = json.loads(config_path.read_text(encoding="utf-8"))
    base_dir = config_path.parent

    original_root = expand_path(raw.get("original_root", "~/dev/edu/repos/stem"), base_dir)
    configured_workspace = os.environ.get("STEM_WORKSPACE_ROOT") or raw.get(
        "workspace_root", "~/dev/workspace/stem"
    )
    workspace_root = expand_path(str(configured_workspace), base_dir)
    legacy_workspace_root = expand_path(
        str(raw.get("legacy_workspace_root", "~/dev/workspace/rv")), base_dir
    )
    if (
        "STEM_WORKSPACE_ROOT" not in os.environ
        and not workspace_root.exists()
        and legacy_workspace_root.exists()
    ):
        workspace_root = legacy_workspace_root
    git_raw = raw.get("git", {})
    base_url = git_raw.get("base_url", "http://77.90.8.171:3001").rstrip("/")
    original_series_root = expand_path(raw.get("original_series_root", str(original_root / "series")), base_dir)
    series_root = expand_path(raw.get("series_root", str(workspace_root / "series")), base_dir)
    configured_socket_root = os.environ.get("STEM_SOCKET_ROOT") or raw.get("socket_root")
    socket_root = (
        expand_path(str(configured_socket_root), base_dir)
        if configured_socket_root
        else default_runtime_socket_root(workspace_root)
    )
    token_path = expand_path(raw.get("token_file", str(workspace_root / "tokens" / "tokens.json")), base_dir)
    workspace_info_path = expand_path(
        raw.get("workspace_info_path", str(workspace_root / "meta" / "workspace-info" / "workspace.json")),
        base_dir,
    )

    candidate_values = raw.get("tools_root_candidates")
    if not candidate_values:
        candidate_values = [
            raw.get("tools_root", str(workspace_root / "tools" / "rv32i-hazard3-student-env")),
            str(workspace_root / "tools" / "rv32i-hazard3-env"),
            str(Path.home() / "dev" / "workspace" / "stem" / "tools" / "rv32i-hazard3-student-env"),
            str(Path.home() / "dev" / "workspace" / "rv" / "tools" / "rv32i-hazard3-student-env"),
            str(original_root / "rv32i-hazard3-student-env"),
            str(original_root / "rv32i-hazard3-env"),
        ]
    tools_root_candidates = [expand_path(value, base_dir) for value in candidate_values]
    adjacent_tools = (base_dir.parent / "rv32i-hazard3-student-env").resolve()
    if adjacent_tools.is_dir() and adjacent_tools not in tools_root_candidates:
        tools_root_candidates.insert(0, adjacent_tools)

    tools_root = next((path for path in tools_root_candidates if path.exists()), tools_root_candidates[0])
    env_tool_url = raw.get("env_tool_url", f"{base_url}/edu-tools/rv32i-hazard3-student-env.git")
    defaults = raw.get("defaults", {})
    workspace_info_url = raw.get("workspace_info_url", f"{base_url}/edu-workspace/workspace-info.git")
    git = GitConfig(
        base_url=base_url,
        source_org=git_raw.get("source_org", "edu-inf"),
        answer_org=git_raw.get("answer_org", "c2025-1a-inf"),
        source_remote=git_raw.get("source_remote", "r1"),
        answer_remote=git_raw.get("answer_remote", "r1a"),
        origin_remote=git_raw.get("origin_remote", "origin"),
        fallback_branch=git_raw.get("fallback_branch", "build"),
    )
    workspace_info = json.loads(workspace_info_path.read_text(encoding="utf-8")) if workspace_info_path.is_file() else None

    return WorkspaceConfig(
        config_path=config_path,
        original_root=original_root,
        original_series_root=original_series_root,
        workspace_root=workspace_root,
        workspace_info_path=workspace_info_path,
        workspace_info_url=workspace_info_url,
        workspace_info=workspace_info,
        series_root=series_root,
        socket_root=socket_root,
        token_path=token_path,
        tools_root_candidates=tools_root_candidates,
        tools_root=tools_root,
        env_tool_url=env_tool_url,
        defaults=defaults,
        git=git,
    )


def write_workspace_config(config: WorkspaceConfig, raw_config: dict) -> None:
    config.config_path.parent.mkdir(parents=True, exist_ok=True)
    temp_path = config.config_path.with_suffix(config.config_path.suffix + ".tmp")
    temp_path.write_text(json.dumps(raw_config, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    temp_path.replace(config.config_path)


def update_config_defaults(config: WorkspaceConfig, updates: dict[str, str]) -> dict[str, str]:
    raw_config = json.loads(config.config_path.read_text(encoding="utf-8"))
    defaults = raw_config.get("defaults")
    if not isinstance(defaults, dict):
        defaults = {}
        raw_config["defaults"] = defaults

    result: dict[str, str] = {}
    changed = False
    for key, value in updates.items():
        old_value = str(defaults.get(key, ""))
        if old_value != value:
            defaults[key] = value
            changed = True
        config.defaults[key] = value
        result[f"old_{key}"] = old_value
        result[key] = value

    if changed:
        write_workspace_config(config, raw_config)
        result["status"] = "updated"
    else:
        result["status"] = "unchanged"
    result["config"] = str(config.config_path)
    return result


def workspace_info_root(config: WorkspaceConfig) -> Path:
    return config.workspace_info_path.parent


def read_workspace_info_file(config: WorkspaceConfig, raw_path: str) -> dict:
    info_path = expand_path(raw_path, workspace_info_root(config))
    return json.loads(info_path.read_text(encoding="utf-8"))


def workspace_series_manifests(config: WorkspaceConfig) -> dict[str, dict]:
    if not config.workspace_info:
        return {}

    manifests: dict[str, dict] = {}
    for entry in config.workspace_info.get("series", []):
        if not isinstance(entry, dict):
            continue
        series_id = str(entry.get("id", "")).strip()
        if not series_id:
            continue
        manifest = dict(entry)
        file_name = entry.get("file")
        if isinstance(file_name, str) and file_name:
            manifest = read_workspace_info_file(config, file_name)
            manifest.setdefault("id", series_id)
        manifests[series_id] = manifest
    return manifests


def workspace_series_manifest(config: WorkspaceConfig, series_name: str) -> dict | None:
    return workspace_series_manifests(config).get(series_name)


def workspace_organizations(config: WorkspaceConfig) -> dict[str, dict]:
    if not config.workspace_info:
        return {}

    organizations: dict[str, dict] = {}
    for entry in config.workspace_info.get("organizations", []):
        if not isinstance(entry, dict):
            continue
        organization_id = str(entry.get("id", "")).strip()
        if organization_id:
            organizations[organization_id] = entry
    return organizations


def audit_workspace_catalog(config: WorkspaceConfig) -> list[tuple[str, str, str]]:
    """Validate the catalog-to-Gitea ownership contract without network access."""
    checks: list[tuple[str, str, str]] = []
    if not config.workspace_info:
        return [("catalog", "FAIL", f"missing {config.workspace_info_path}")]

    raw_organizations = config.workspace_info.get("organizations", [])
    organizations = workspace_organizations(config)
    if not isinstance(raw_organizations, list) or not raw_organizations:
        checks.append(("organizations", "FAIL", "workspace.json must declare the organization registry"))
    else:
        raw_ids = [str(entry.get("id", "")).strip() for entry in raw_organizations if isinstance(entry, dict)]
        duplicates = sorted({value for value in raw_ids if value and raw_ids.count(value) > 1})
        if duplicates:
            checks.append(("organizations", "FAIL", f"duplicate ids: {', '.join(duplicates)}"))
        else:
            checks.append(("organizations", "ok", f"{len(organizations)} registered"))

    control_plane = organizations.get("edu")
    if not control_plane or control_plane.get("role") != "control-plane":
        checks.append(("control-plane", "FAIL", "organization 'edu' must have role 'control-plane'"))
    else:
        checks.append(("control-plane", "ok", "edu"))

    manifests = workspace_series_manifests(config)
    if not manifests:
        checks.append(("series", "FAIL", "no operational series declared"))
        return checks

    physical_workspace_dirs: dict[str, list[str]] = {}
    for series_id, manifest in manifests.items():
        source_org = manifest.get("source_org")
        if not isinstance(source_org, str) or not source_org.strip():
            checks.append((f"series:{series_id}", "FAIL", "missing explicit source_org"))
            continue
        source_org = source_org.strip()
        organization = organizations.get(source_org)
        if organization is None:
            checks.append((f"series:{series_id}", "FAIL", f"source_org '{source_org}' is not registered"))
        else:
            assigned_series = organization.get("series", [])
            if not isinstance(assigned_series, list) or series_id not in assigned_series:
                checks.append(
                    (f"series:{series_id}", "FAIL", f"organization '{source_org}' does not assign this series")
                )
            else:
                checks.append((f"series:{series_id}", "ok", f"source_org={source_org}"))

        expected_org = f"edu-{series_id.replace('_', '-')}"
        if source_org != expected_org and not bool(manifest.get("legacy")):
            checks.append(
                (f"naming:{series_id}", "WARN", f"expected '{expected_org}', declare legacy=true if intentional")
            )

        raw_cards = manifest.get("cards", [])
        cards = [card for card in raw_cards if isinstance(card, dict)] if isinstance(raw_cards, list) else []
        card_ids = [str(card.get("id", "")).strip() for card in cards]
        repos = [str(card.get("repo", "")).strip() for card in cards]
        duplicate_ids = sorted({value for value in card_ids if value and card_ids.count(value) > 1})
        duplicate_repos = sorted({value for value in repos if value and repos.count(value) > 1})
        missing = sum(not card_id or not repo for card_id, repo in zip(card_ids, repos))
        if duplicate_ids or duplicate_repos or missing:
            details = []
            if duplicate_ids:
                details.append(f"duplicate card ids: {', '.join(duplicate_ids)}")
            if duplicate_repos:
                details.append(f"duplicate repos: {', '.join(duplicate_repos)}")
            if missing:
                details.append(f"{missing} cards missing id/repo")
            checks.append((f"cards:{series_id}", "FAIL", "; ".join(details)))
        else:
            checks.append((f"cards:{series_id}", "ok", f"{len(cards)} cards"))

        try:
            physical_dir = series_path_component(config, series_id, "workspace_dir")
        except SystemExit as error:
            checks.append((f"path:{series_id}", "FAIL", str(error)))
        else:
            physical_workspace_dirs.setdefault(physical_dir, []).append(series_id)

    for physical_dir, series_ids in physical_workspace_dirs.items():
        if len(series_ids) > 1:
            checks.append(
                ("workspace-dir", "WARN", f"'{physical_dir}' is shared by: {', '.join(series_ids)}")
            )
    return checks


def series_path_component(config: WorkspaceConfig, series_name: str, field: str) -> str:
    """Return a safe physical directory name for one logical series."""
    manifest = workspace_series_manifest(config, series_name)
    raw_value = manifest.get(field, series_name) if manifest else series_name
    value = str(raw_value).strip()
    if not value or value in {".", ".."} or Path(value).name != value:
        raise SystemExit(f"Invalid {field} for series '{series_name}': {raw_value!r}")
    return value


def workspace_series_path(config: WorkspaceConfig, series_name: str) -> Path:
    return config.series_root / series_path_component(config, series_name, "workspace_dir")


def source_series_path(config: WorkspaceConfig, series_name: str) -> Path:
    return config.original_series_root / series_path_component(config, series_name, "source_dir")


def series_exists(config: WorkspaceConfig, series_name: str) -> bool:
    if workspace_series_manifest(config, series_name) is not None:
        return True
    return workspace_series_path(config, series_name).is_dir() or source_series_path(config, series_name).is_dir()


def default_card_for_series(config: WorkspaceConfig, series_name: str) -> str | None:
    manifest = workspace_series_manifest(config, series_name)
    if manifest and isinstance(manifest.get("default_card"), str):
        return manifest["default_card"]
    return config.defaults.get("card")


def config_for_card(config: WorkspaceConfig, card_info: CardInfo) -> WorkspaceConfig:
    return replace(
        config,
        git=replace(
            config.git,
            source_org=card_info.source_org,
            answer_org=card_info.answer_org,
            source_remote=card_info.source_remote,
            answer_remote=card_info.answer_remote,
            fallback_branch=card_info.fallback_branch,
        ),
    )


def card_infos_from_manifest(config: WorkspaceConfig, series_name: str) -> list[CardInfo]:
    manifest = workspace_series_manifest(config, series_name)
    if not manifest:
        return []

    raw_source_org = manifest.get("source_org")
    if not isinstance(raw_source_org, str) or not raw_source_org.strip():
        raise SystemExit(
            f"Series '{series_name}' must declare source_org explicitly in workspace-info; "
            "the global git.source_org is only a legacy fallback for directory-only workspaces."
        )
    source_org = raw_source_org.strip()
    answer_org = str(manifest.get("answer_org", config.git.answer_org))
    source_remote = str(manifest.get("source_remote", config.git.source_remote))
    answer_remote = str(manifest.get("answer_remote", config.git.answer_remote))
    fallback_branch = str(manifest.get("fallback_branch", config.git.fallback_branch))
    cards = []
    for raw_card in manifest.get("cards", []):
        if not isinstance(raw_card, dict):
            continue
        card_id = str(raw_card.get("id", "")).strip()
        if not card_id:
            continue
        repo_name = str(raw_card.get("repo", card_id)).strip()
        branch_name = str(raw_card.get("branch", fallback_branch)).strip() or fallback_branch
        title = str(raw_card.get("title", "")).strip()
        source_path = source_series_path(config, series_name) / repo_name
        cards.append(
            CardInfo(
                series_id=series_name,
                card_id=card_id,
                repo=repo_name,
                branch=branch_name,
                title=title,
                source_org=source_org,
                answer_org=answer_org,
                source_remote=source_remote,
                answer_remote=answer_remote,
                fallback_branch=fallback_branch,
                source_path=source_path if source_path.is_dir() else None,
            )
        )
    return cards


def card_infos_from_source_dirs(config: WorkspaceConfig, series_name: str) -> list[CardInfo]:
    return [
        CardInfo(
            series_id=series_name,
            card_id=card_label_from_repo_name(path.name),
            repo=path.name,
            branch=git_current_branch(config, path),
            title=read_card_title(path),
            source_org=config.git.source_org,
            answer_org=config.git.answer_org,
            source_remote=config.git.source_remote,
            answer_remote=config.git.answer_remote,
            fallback_branch=config.git.fallback_branch,
            source_path=path,
        )
        for path in source_card_directories(config, series_name)
    ]


def source_card_infos(config: WorkspaceConfig, series_name: str) -> list[CardInfo]:
    manifest_cards = card_infos_from_manifest(config, series_name)
    if manifest_cards:
        return manifest_cards
    return card_infos_from_source_dirs(config, series_name)


def series_directories(config: WorkspaceConfig) -> list[Path]:
    config.series_root.mkdir(parents=True, exist_ok=True)
    return sorted(path for path in config.series_root.iterdir() if path.is_dir() and not path.name.startswith("."))


def card_directories(series_path: Path) -> list[Path]:
    return sorted(path for path in series_path.iterdir() if path.is_dir() and not path.name.startswith("."))


def source_series_directories(config: WorkspaceConfig) -> list[Path]:
    if not config.original_series_root.is_dir():
        return []
    return sorted(path for path in config.original_series_root.iterdir() if path.is_dir() and not path.name.startswith("."))


def source_card_directories(config: WorkspaceConfig, series_name: str) -> list[Path]:
    series_path = source_series_path(config, series_name)
    if not series_path.is_dir():
        return []
    return sorted(path for path in series_path.iterdir() if path.is_dir() and not path.name.startswith("."))


def card_label_from_repo_name(repo_name: str) -> str:
    if repo_name.startswith("lab-"):
        repo_name = repo_name[4:]
    marker = "-strlen-"
    if marker in repo_name:
        return repo_name.split(marker, 1)[1].split("-", 1)[0]
    return repo_name


def card_info_matches(card_info: CardInfo, normalized: str) -> bool:
    candidates = {
        str(card_info.card_id).lower(),
        str(card_info.repo).lower(),
        card_label_from_repo_name(card_info.repo).lower(),
    }
    return normalized in candidates or any(normalized in candidate for candidate in candidates)


def resolve_card_info(config: WorkspaceConfig, series_name: str, card_name: str) -> CardInfo:
    cards = source_card_infos(config, series_name)
    if not cards:
        source_hint = source_series_path(config, series_name)
        if config.workspace_info_path.is_file():
            source_hint = config.workspace_info_path
        raise SystemExit(f"Missing source series: {source_hint}")

    normalized = slug_value(card_name, "card")
    matches = [card_info for card_info in cards if card_info_matches(card_info, normalized)]
    if not matches:
        available = ", ".join(card.card_id for card in cards)
        raise SystemExit(f"Missing card '{card_name}' in series '{series_name}'. Available: {available}")
    if len(matches) > 1:
        available = ", ".join(card.repo for card in matches)
        raise SystemExit(f"Ambiguous card '{card_name}' in series '{series_name}'. Matches: {available}")
    return matches[0]


def resolve_source_card_path(config: WorkspaceConfig, series_name: str, card_name: str) -> Path:
    manifest_card = resolve_card_info(config, series_name, card_name)
    if manifest_card.source_path:
        return manifest_card.source_path
    return source_series_path(config, series_name) / manifest_card.repo


def resolve_source_card_path_from_dirs(config: WorkspaceConfig, series_name: str, card_name: str) -> Path:
    cards = source_card_directories(config, series_name)
    if not cards:
        raise SystemExit(f"Missing source series: {source_series_path(config, series_name)}")

    normalized = slug_value(card_name, "card")
    matches = []
    for card_path in cards:
        repo_name = card_path.name
        label = card_label_from_repo_name(repo_name)
        candidates = {repo_name, label}
        if normalized in candidates or any(normalized in candidate for candidate in candidates):
            matches.append(card_path)

    if not matches:
        available = ", ".join(card_label_from_repo_name(path.name) for path in cards)
        raise SystemExit(f"Missing card '{card_name}' in series '{series_name}'. Available: {available}")
    if len(matches) > 1:
        available = ", ".join(path.name for path in matches)
        raise SystemExit(f"Ambiguous card '{card_name}' in series '{series_name}'. Matches: {available}")
    return matches[0]


def read_card_title(card_path: Path) -> str:
    readme_path = card_path / "README.md"
    if not readme_path.is_file():
        return ""

    for line in readme_path.read_text(encoding="utf-8").splitlines():
        stripped = line.strip()
        if stripped.startswith("#"):
            return stripped.lstrip("#").strip()
    return ""


def resolve_selector(
    config: WorkspaceConfig,
    series_arg: str | None,
    card_arg: str | None,
) -> tuple[str, str]:
    default_series = config.defaults.get("series")
    default_card = config.defaults.get("card")

    if series_arg and "/" in series_arg:
        if card_arg:
            raise SystemExit("Pass either '<series> <card>' or '<series/card>', not both forms at once.")
        series_value, card_value = series_arg.split("/", 1)
        return series_value, card_value

    if series_arg and not card_arg:
        if series_exists(config, series_arg):
            series_default_card = default_card_for_series(config, series_arg)
            if not series_default_card:
                raise SystemExit("Missing default card in config.")
            return series_arg, series_default_card
        if default_series and (workspace_series_path(config, default_series) / series_arg).is_dir():
            return default_series, series_arg
        if default_series:
            try:
                source_card_path = resolve_source_card_path(config, default_series, series_arg)
                if workspace_card_path(config, default_series, source_card_path).is_dir():
                    return default_series, series_arg
            except SystemExit:
                pass

    series_value = series_arg or default_series
    card_value = card_arg or default_card
    if not series_value or not card_value:
        raise SystemExit("Missing series/card selector and no defaults are configured.")
    return series_value, card_value


def resolve_card_path(config: WorkspaceConfig, series_name: str, card_name: str) -> Path:
    card_path = workspace_series_path(config, series_name) / card_name
    if card_path.is_dir():
        return card_path
    card_info = resolve_card_info(config, series_name, card_name)
    resolved_path = workspace_card_path_for_info(config, card_info)
    if resolved_path.is_dir():
        return resolved_path
    raise SystemExit(f"Missing card path: {resolved_path}")


def print_series(config: WorkspaceConfig) -> None:
    config.series_root.mkdir(parents=True, exist_ok=True)
    manifest_by_name = workspace_series_manifests(config)
    if manifest_by_name:
        for series_name in manifest_by_name:
            workspace_path = workspace_series_path(config, series_name)
            cards = source_card_infos(config, series_name)
            workspace_count = sum((workspace_path / card.repo).is_dir() for card in cards)
            print(f"{series_name}\t{len(cards)}\t{workspace_count}\t{workspace_path if workspace_path.is_dir() else ''}")
        return

    source_by_name = {path.name: path for path in source_series_directories(config)}
    workspace_by_name = {path.name: path for path in series_directories(config)}
    for series_name in sorted(set(manifest_by_name) | set(source_by_name) | set(workspace_by_name)):
        manifest = manifest_by_name.get(series_name)
        workspace_path = workspace_by_name.get(series_name)
        source_count = len(source_card_infos(config, series_name)) if (manifest or series_name in source_by_name) else 0
        workspace_count = len(card_directories(workspace_path)) if workspace_path else 0
        print(f"{series_name}\t{source_count}\t{workspace_count}\t{workspace_path or ''}")


def print_cards(config: WorkspaceConfig, series_arg: str | None) -> None:
    series_name = series_arg or config.defaults.get("series")
    if not series_name:
        raise SystemExit("Missing series name and no default series is configured.")

    config.series_root.mkdir(parents=True, exist_ok=True)
    source_cards = {card.repo: card for card in source_card_infos(config, series_name)}
    series_path = workspace_series_path(config, series_name)
    if workspace_series_manifest(config, series_name):
        workspace_cards = {
            card.repo: series_path / card.repo
            for card in source_cards.values()
            if (series_path / card.repo).is_dir()
        }
    else:
        workspace_cards = {path.name: path for path in card_directories(series_path)} if series_path.is_dir() else {}
    if not source_cards and not workspace_cards:
        raise SystemExit(f"Missing series: {series_name}")

    for repo_name in sorted(set(source_cards) | set(workspace_cards)):
        source_card = source_cards.get(repo_name)
        workspace_path = workspace_cards.get(repo_name)
        title = read_card_title(workspace_path) if workspace_path else ""
        if not title and source_card:
            title = source_card.title
        if not title and source_card and source_card.source_path:
            title = read_card_title(source_card.source_path)
        label = source_card.card_id if source_card else card_label_from_repo_name(repo_name)
        status = "workspace" if workspace_path else "source"
        if title:
            print(f"{label}\t{repo_name}\t{status}\t{title}")
        else:
            print(f"{label}\t{repo_name}\t{status}")


def slug_value(raw_value: str, field_name: str) -> str:
    slug = re.sub(r"[^a-z0-9._-]+", "-", raw_value.strip().lower()).strip("-")
    if not slug:
        raise SystemExit(f"Empty {field_name} after slug conversion: {raw_value!r}")
    return slug


def branch_value(raw_value: str, field_name: str = "branch") -> str:
    branch_name = re.sub(r"[^A-Za-z0-9._-]+", "-", raw_value.strip()).strip("-")
    if not branch_name:
        raise SystemExit(f"Empty {field_name} after branch conversion: {raw_value!r}")
    return branch_name


def git_capture(repo_path: Path, git_args: list[str]) -> str | None:
    result = subprocess.run(
        ["git", "-C", str(repo_path)] + git_args,
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        return None
    return result.stdout.strip()


def git_repo_root(repo_path: Path) -> Path | None:
    repo_root = git_capture(repo_path, ["rev-parse", "--show-toplevel"])
    if not repo_root:
        return None
    return Path(repo_root)


def default_port_for_scheme(scheme: str) -> int | None:
    if scheme == "http":
        return 80
    if scheme == "https":
        return 443
    return None


def endpoint_from_split(split_url) -> tuple[str, int | None]:
    host_name = split_url.hostname
    if not host_name:
        raise SystemExit(f"Missing hostname in URL: {split_url.geturl()}")

    port_value = split_url.port or default_port_for_scheme(split_url.scheme)
    default_port = default_port_for_scheme(split_url.scheme)
    if port_value is not None and port_value != default_port:
        endpoint = f"{split_url.scheme}://{host_name}:{port_value}"
    else:
        endpoint = f"{split_url.scheme}://{host_name}"
    return endpoint, port_value


def server_type_from_split(config: WorkspaceConfig, split_url) -> str:
    endpoint, _ = endpoint_from_split(split_url)
    base_split = urlsplit(config.git.base_url)
    base_endpoint, _ = endpoint_from_split(base_split)
    if endpoint == base_endpoint:
        return "gitea"

    host_name = (split_url.hostname or "").lower()
    if host_name == "github.com" or host_name.endswith(".github.com"):
        return "github"
    if "gitlab" in host_name:
        return "gitlab"
    if "gitea" in host_name:
        return "gitea"
    return "unknown"


def repo_parts_from_url(remote_url: str) -> tuple[str, str]:
    split_url = urlsplit(remote_url)
    repo_path = unquote(split_url.path).strip("/")
    if repo_path.endswith(".git"):
        repo_path = repo_path[:-4]
    path_parts = [part for part in repo_path.split("/") if part]
    if len(path_parts) < 2:
        return "", ""
    return path_parts[0], path_parts[1]


def server_info_from_url(config: WorkspaceConfig, remote_url: str) -> dict | None:
    split_url = urlsplit(remote_url)
    if split_url.scheme not in {"http", "https"}:
        return None
    if split_url.hostname is None:
        return None

    endpoint, port_value = endpoint_from_split(split_url)
    org_name, repo_name = repo_parts_from_url(remote_url)
    return {
        "endpoint": endpoint,
        "type": server_type_from_split(config, split_url),
        "scheme": split_url.scheme,
        "host": split_url.hostname,
        "port": port_value,
        "org": org_name,
        "repo": repo_name,
    }


def empty_token_store() -> dict:
    return {"version": 3, "tokens": []}


def next_token_name(tokens: dict[str, str]) -> str:
    index = 1
    while f"t{index}" in tokens:
        index += 1
    return f"t{index}"


def server_record_from_info(server_info: dict) -> dict:
    return {
        "type": str(server_info.get("type", "unknown")),
        "endpoint": str(server_info.get("endpoint", "")),
        "scheme": str(server_info.get("scheme", "")),
        "host": str(server_info.get("host", "")),
        "port": server_info.get("port"),
    }


def normalize_server_record(config: WorkspaceConfig, raw_server: dict, endpoint: str | None = None) -> dict | None:
    if not isinstance(raw_server, dict):
        return None

    endpoint_value = str(raw_server.get("endpoint") or endpoint or "")
    if not endpoint_value:
        return None

    split_endpoint = urlsplit(endpoint_value)
    host_value = raw_server.get("host") or split_endpoint.hostname or ""
    scheme_value = raw_server.get("scheme") or split_endpoint.scheme
    port_value = raw_server.get("port")
    if port_value is None and split_endpoint.port is not None:
        port_value = split_endpoint.port

    server_type = raw_server.get("type") or "unknown"
    if server_type == "unknown" and scheme_value and host_value:
        server_url = endpoint_value
        server_type = server_type_from_split(config, urlsplit(server_url))

    return {
        "type": str(server_type),
        "endpoint": endpoint_value,
        "scheme": str(scheme_value),
        "host": str(host_value),
        "port": port_value,
    }


def normalize_token_entry(raw_token_entry) -> str | dict | None:
    if isinstance(raw_token_entry, str):
        return raw_token_entry if raw_token_entry else None
    if not isinstance(raw_token_entry, dict):
        return None

    token_value = raw_token_entry.get("value")
    if not isinstance(token_value, str) or not token_value:
        return None

    token_entry = {"value": token_value}
    expires_at = raw_token_entry.get("expires_at")
    if isinstance(expires_at, str) and expires_at:
        token_entry["expires_at"] = expires_at
    for field_name in ["remote", "org", "repo"]:
        field_value = raw_token_entry.get(field_name)
        if isinstance(field_value, str) and field_value:
            token_entry[field_name] = field_value
    return token_entry


def token_entry_value(token_entry) -> str:
    if isinstance(token_entry, str):
        return token_entry
    if isinstance(token_entry, dict) and isinstance(token_entry.get("value"), str):
        return token_entry["value"]
    return ""


def token_entry_expires_at(token_entry) -> str:
    if isinstance(token_entry, dict) and isinstance(token_entry.get("expires_at"), str):
        return token_entry["expires_at"]
    return ""


def token_entry_field(token_entry, field_name: str) -> str:
    if isinstance(token_entry, dict) and isinstance(token_entry.get(field_name), str):
        return token_entry[field_name]
    return ""


def token_value_exists(tokens: dict, token_value: str) -> bool:
    return any(token_entry_value(existing_token) == token_value for existing_token in tokens.values())


def normalize_remote_record(raw_remote: dict) -> dict | None:
    if not isinstance(raw_remote, dict):
        return None
    remote_name = raw_remote.get("name") or raw_remote.get("remote")
    if not isinstance(remote_name, str) or not remote_name:
        return None

    remote_record = {"name": remote_name, "org": "", "repo": ""}
    for field_name in ["org", "repo"]:
        field_value = raw_remote.get(field_name)
        if isinstance(field_value, str) and field_value:
            remote_record[field_name] = field_value

    org_perm = normalize_permission_map(raw_remote.get("org_perm"), ORG_PERMISSION_FIELDS, "+?!-")
    if org_perm is not None:
        remote_record["org_perm"] = org_perm
    repo_perm = normalize_permission_map(raw_remote.get("repo_perm"), REPO_PERMISSION_FIELDS, "+?!-")
    if repo_perm is not None:
        remote_record["repo_perm"] = repo_perm
    return remote_record


def valid_mask(raw_value: str, allowed: str, width: int) -> bool:
    return len(raw_value) == width and all(char in allowed for char in raw_value)


def permission_map_from_mask(
    raw_value: str,
    fields: list[tuple[str, str]],
    allowed: str,
) -> dict[str, str] | None:
    if not valid_mask(raw_value, allowed, len(fields)):
        return None
    return {key: value for (key, _), value in zip(fields, raw_value)}


def normalize_permission_map(
    raw_value,
    fields: list[tuple[str, str]],
    allowed: str,
    unknown: str = "?",
) -> dict[str, str] | None:
    if isinstance(raw_value, str):
        return permission_map_from_mask(raw_value, fields, allowed)
    if not isinstance(raw_value, dict):
        return None

    normalized: dict[str, str] = {}
    found = False
    for key, long_name in fields:
        field_value = raw_value.get(key, raw_value.get(long_name))
        if isinstance(field_value, str) and len(field_value) == 1 and field_value in allowed:
            normalized[key] = field_value
            found = True
        else:
            normalized[key] = unknown
    return normalized if found else None


def permission_map_to_mask(
    raw_value,
    fields: list[tuple[str, str]],
    allowed: str,
    unknown: str = "?",
) -> str:
    if isinstance(raw_value, str) and valid_mask(raw_value, allowed, len(fields)):
        return raw_value
    if not isinstance(raw_value, dict):
        return ""

    values = []
    for key, long_name in fields:
        field_value = raw_value.get(key, raw_value.get(long_name))
        if isinstance(field_value, str) and len(field_value) == 1 and field_value in allowed:
            values.append(field_value)
        else:
            values.append(unknown)
    return "".join(values)


def normalize_flat_token(
    config: WorkspaceConfig,
    raw_token: dict,
    remote_id: str,
    token_value: str,
    user_name: str,
    org_name: str,
    repo_name: str,
    org_perm=None,
    repo_perm=None,
) -> dict | None:
    server_record = normalize_server_record(config, raw_token.get("server", {}))
    if server_record is None:
        return None

    token_record = {
        "id": remote_id,
        "value": token_value,
        "server": server_record,
        "org": org_name,
        "repo": repo_name,
    }
    if user_name:
        token_record["user"] = user_name
    for field_name in ["valid", "expires_at"]:
        field_value = raw_token.get(field_name)
        if isinstance(field_value, str) and field_value:
            token_record[field_name] = field_value

    scope_record = normalize_permission_map(raw_token.get("scope"), SCOPE_FIELDS, "rw?!-")
    if scope_record is not None:
        token_record["scope"] = scope_record

    org_record = normalize_permission_map(org_perm if org_perm is not None else raw_token.get("org_perm"), ORG_PERMISSION_FIELDS, "+?!-")
    if org_record is not None:
        token_record["org_perm"] = org_record
    repo_record = normalize_permission_map(repo_perm if repo_perm is not None else raw_token.get("repo_perm"), REPO_PERMISSION_FIELDS, "+?!-")
    if repo_record is not None:
        token_record["repo_perm"] = repo_record
    return token_record


def normalize_v3_tokens(config: WorkspaceConfig, raw_token: dict) -> list[dict]:
    if not isinstance(raw_token, dict):
        return []

    remote_id = raw_token.get("id") or raw_token.get("remote")
    legacy_token_id = raw_token.get("token_id") or raw_token.get("token_ref")
    token_value = raw_token.get("value")
    if token_value is None:
        token_value = raw_token.get("token")
    if not isinstance(token_value, str):
        return []
    user_name = raw_token.get("user")
    if not isinstance(user_name, str):
        user_name = ""

    if isinstance(remote_id, str) and remote_id:
        token_record = normalize_flat_token(
            config,
            raw_token,
            remote_id,
            token_value,
            user_name,
            str(raw_token.get("org", "")),
            str(raw_token.get("repo", "")),
        )
        return [token_record] if token_record is not None else []

    raw_remotes = raw_token.get("remotes", [])
    if isinstance(raw_remotes, dict):
        raw_remotes = [
            {"name": remote_name, **remote_value}
            for remote_name, remote_value in raw_remotes.items()
            if isinstance(remote_value, dict)
        ]
    if not isinstance(raw_remotes, list):
        raw_remotes = []

    normalized_records: list[dict] = []
    if isinstance(raw_remotes, list):
        for raw_remote in raw_remotes:
            remote_record = normalize_remote_record(raw_remote)
            if remote_record is None:
                continue
            legacy_user_name = user_name
            if not legacy_user_name and isinstance(legacy_token_id, str):
                legacy_user_name = legacy_token_id
            token_record = normalize_flat_token(
                config,
                raw_token,
                remote_record["name"],
                token_value,
                legacy_user_name,
                remote_record.get("org", ""),
                remote_record.get("repo", ""),
                remote_record.get("org_perm"),
                remote_record.get("repo_perm"),
            )
            if token_record is not None:
                normalized_records.append(token_record)

    if not normalized_records and isinstance(legacy_token_id, str) and legacy_token_id:
        token_record = normalize_flat_token(
            config,
            raw_token,
            legacy_token_id,
            token_value,
            user_name,
            str(raw_token.get("org", "")),
            str(raw_token.get("repo", "")),
        )
        if token_record is not None:
            normalized_records.append(token_record)
    return normalized_records


def append_token_record(token_data: dict, token_record: dict) -> None:
    tokens = token_data.setdefault("tokens", [])
    token_key = (
        token_record.get("server", {}).get("endpoint", ""),
        token_record.get("id", ""),
    )
    for existing_record in tokens:
        existing_key = (
            existing_record.get("server", {}).get("endpoint", ""),
            existing_record.get("id", ""),
        )
        if existing_key == token_key:
            existing_record.update(token_record)
            return
    tokens.append(token_record)


def parse_datetime(raw_value: str) -> dt.datetime | None:
    if not raw_value:
        return None
    try:
        normalized_value = raw_value.replace("Z", "+00:00")
        return dt.datetime.fromisoformat(normalized_value)
    except ValueError:
        return None


def is_expired(raw_value: str) -> bool:
    expires_at = parse_datetime(raw_value)
    if expires_at is None:
        return False
    if expires_at.tzinfo is None:
        now = dt.datetime.now()
    else:
        now = dt.datetime.now(expires_at.tzinfo)
    return expires_at <= now


def valid_label(api_accepts_token: bool, expires_at: str) -> str:
    if not api_accepts_token or (expires_at and is_expired(expires_at)):
        return "invalid"
    if expires_at:
        return expires_at
    return "forever"


def copy_legacy_users(token_data: dict, server_info: dict, raw_users: dict) -> dict:
    for user_name, user_entry in raw_users.items():
        if not isinstance(user_entry, dict):
            continue
        raw_tokens = user_entry.get("tokens", {})
        if not isinstance(raw_tokens, dict):
            continue

        for token_name, raw_token_entry in raw_tokens.items():
            token_entry = normalize_token_entry(raw_token_entry)
            if token_entry is None:
                continue
            token_value = token_entry_value(token_entry)
            remote_name = token_entry_field(token_entry, "remote") or str(token_name)

            token_record = {
                "id": remote_name,
                "value": token_value,
                "server": server_record_from_info(server_info),
                "user": str(user_name),
                "org": token_entry_field(token_entry, "org"),
                "repo": token_entry_field(token_entry, "repo"),
            }
            expires_at = token_entry_expires_at(token_entry)
            if expires_at:
                token_record["expires_at"] = expires_at

            append_token_record(token_data, token_record)

    return token_data


def normalize_token_store(config: WorkspaceConfig, raw_data: dict) -> dict:
    if raw_data.get("version") == 3 and isinstance(raw_data.get("tokens"), list):
        token_data = empty_token_store()
        for raw_token in raw_data["tokens"]:
            for token_record in normalize_v3_tokens(config, raw_token):
                append_token_record(token_data, token_record)
        return token_data

    if "servers" in raw_data and isinstance(raw_data["servers"], dict):
        token_data = empty_token_store()
        for endpoint, server_entry in raw_data["servers"].items():
            if not isinstance(server_entry, dict):
                continue

            base_server_info = {
                "endpoint": str(endpoint),
                "type": server_entry.get("type", "unknown"),
                "scheme": server_entry.get("scheme", ""),
                "host": server_entry.get("host", ""),
                "port": server_entry.get("port"),
            }
            copy_legacy_users(token_data, base_server_info, server_entry.get("users", {}))
        return token_data

    token_data = empty_token_store()
    if "users" in raw_data and isinstance(raw_data["users"], dict):
        base_server_info = server_info_from_url(config, config.git.base_url)
        if base_server_info is None:
            raise SystemExit(f"Invalid git base URL: {config.git.base_url}")
        copy_legacy_users(token_data, base_server_info, raw_data["users"])
    return token_data


def load_token_store(config: WorkspaceConfig, write_normalized: bool = True) -> dict:
    if not config.token_path.exists():
        return empty_token_store()

    raw_data = json.loads(config.token_path.read_text(encoding="utf-8"))
    if not isinstance(raw_data, dict):
        raise SystemExit(f"Invalid token store format: {config.token_path}")
    normalized_data = normalize_token_store(config, raw_data)
    if write_normalized and raw_data != normalized_data:
        write_token_store(config, normalized_data)
    return normalized_data


def write_token_store(config: WorkspaceConfig, token_data: dict) -> None:
    config.token_path.parent.mkdir(parents=True, exist_ok=True)
    os.chmod(config.token_path.parent, 0o700)

    temp_path = config.token_path.with_suffix(config.token_path.suffix + ".tmp")
    temp_path.write_text(json.dumps(token_data, indent=2) + "\n", encoding="utf-8")
    os.chmod(temp_path, 0o600)
    temp_path.replace(config.token_path)
    os.chmod(config.token_path, 0o600)


def token_server_endpoint(token_record: dict) -> str:
    server_record = token_record.get("server", {})
    if not isinstance(server_record, dict):
        return ""
    return str(server_record.get("endpoint", ""))


def token_record_id(token_record: dict) -> str:
    return str(token_record.get("id", ""))


def find_token_record(token_data: dict, endpoint: str, remote_id: str) -> dict | None:
    for token_record in token_data.get("tokens", []):
        if token_server_endpoint(token_record) == endpoint and token_record_id(token_record) == remote_id:
            return token_record
    return None


def apply_authz_to_token(token_record: dict, authz: dict) -> bool:
    changed = False
    for source_name, target_name in [
        ("user", "user"),
        ("valid", "valid"),
    ]:
        source_value = authz.get(source_name)
        if isinstance(source_value, str) and source_value and token_record.get(target_name) != source_value:
            token_record[target_name] = source_value
            changed = True

    scope_value = authz.get("scope_mask")
    if isinstance(scope_value, str):
        scope_record = permission_map_from_mask(scope_value, SCOPE_FIELDS, "rw?!-")
        if scope_record is not None and token_record.get("scope") != scope_record:
            token_record["scope"] = scope_record
            changed = True

    for source_name, target_name in [
        ("org_mask", "org_perm"),
        ("repo_mask", "repo_perm"),
    ]:
        source_value = authz.get(source_name)
        if not isinstance(source_value, str):
            continue
        fields = ORG_PERMISSION_FIELDS if target_name == "org_perm" else REPO_PERMISSION_FIELDS
        permission_record = permission_map_from_mask(source_value, fields, "+?!-")
        if permission_record is not None and token_record.get(target_name) != permission_record:
            token_record[target_name] = permission_record
            changed = True
    return changed


def unknown_permission_map(fields: list[tuple[str, str]]) -> dict[str, str]:
    return {short_name: "?" for short_name, _ in fields}


def reset_token_api_metadata(token_record: dict) -> bool:
    changed = False
    if token_record.get("valid") != "?":
        token_record["valid"] = "?"
        changed = True
    if token_record.pop("expires_at", None) is not None:
        changed = True

    for field_name, fields in [
        ("scope", SCOPE_FIELDS),
        ("org_perm", ORG_PERMISSION_FIELDS),
        ("repo_perm", REPO_PERMISSION_FIELDS),
    ]:
        unknown_record = unknown_permission_map(fields)
        if token_record.get(field_name) != unknown_record:
            token_record[field_name] = unknown_record
            changed = True
    return changed


def register_token(
    token_data: dict,
    server_info: dict,
    remote_name: str,
    user_name: str,
    token_value: str,
    hydrate: bool = True,
    reset_api_metadata: bool = False,
) -> str:
    endpoint = server_info["endpoint"]
    token_record = find_token_record(token_data, endpoint, remote_name)
    if token_record is None:
        token_record = {
            "id": remote_name,
            "value": token_value,
            "server": server_record_from_info(server_info),
            "user": user_name,
            "org": str(server_info.get("org", "")),
            "repo": str(server_info.get("repo", "")),
        }
        token_data.setdefault("tokens", []).append(token_record)
        result = "added"
    else:
        result = "existing"
        if token_record.get("value") != token_value:
            token_record["value"] = token_value
            result = "updated"
        if token_record.get("user") != user_name:
            token_record["user"] = user_name
            result = "updated"
        if token_record.get("server") != server_record_from_info(server_info):
            token_record["server"] = server_record_from_info(server_info)
            result = "updated"
        for source_name, target_name in [("org", "org"), ("repo", "repo")]:
            source_value = server_info.get(source_name)
            if isinstance(source_value, str) and token_record.get(target_name) != source_value:
                token_record[target_name] = source_value
                result = "updated"

    if reset_api_metadata:
        if reset_token_api_metadata(token_record) and result == "existing":
            result = "updated"

    if hydrate:
        authz = load_gitea_authz(
            endpoint,
            remote_name,
            token_value,
            token_record.get("expires_at", ""),
            token_record.get("org", ""),
            token_record.get("repo", ""),
        )
        if apply_authz_to_token(token_record, authz) and result == "existing":
            result = "updated"
    return result


def remote_urls(repo_path: Path) -> list[str]:
    return [remote_url for _, remote_url in remote_name_urls(repo_path)]


def remote_name_urls(repo_path: Path) -> list[tuple[str, str]]:
    remote_output = git_capture(repo_path, ["remote"])
    if not remote_output:
        return []

    urls: list[tuple[str, str]] = []
    seen: set[tuple[str, str]] = set()
    for remote_name in remote_output.splitlines():
        url_output = git_capture(repo_path, ["remote", "get-url", "--all", remote_name])
        if not url_output:
            continue
        for remote_url in url_output.splitlines():
            key = (remote_name, remote_url)
            if key in seen:
                continue
            seen.add(key)
            urls.append(key)
    return urls


def remote_credentials(remote_url: str) -> tuple[str, str] | None:
    split_url = urlsplit(remote_url)
    if split_url.scheme not in {"http", "https"}:
        return None
    if split_url.username is None or split_url.password is None:
        return None

    return unquote(split_url.username), unquote(split_url.password)


def scan_repo_remotes(config: WorkspaceConfig, repo_path: Path) -> dict:
    repo_root = git_repo_root(repo_path)
    if repo_root is None:
        return {
            "repo_root": None,
            "remotes": 0,
            "urls": 0,
            "scanned": 0,
            "auth_urls": 0,
            "plain_urls": 0,
            "unsupported_urls": 0,
            "found": 0,
            "added": 0,
            "updated": 0,
            "servers": 0,
            "remote_rows": [],
            "found_credentials": [],
            "changed": False,
        }

    found_credentials: list[tuple[int, dict, str, str, str]] = []
    remote_entries = remote_name_urls(repo_root)
    remote_names = {remote_name for remote_name, _ in remote_entries}
    touched_servers: set[str] = set()
    remote_rows: list[dict[str, str]] = []
    auth_urls = 0
    plain_urls = 0
    unsupported_urls = 0
    for remote_name, remote_url in remote_entries:
        server_info = server_info_from_url(config, remote_url)
        if server_info is None:
            unsupported_urls += 1
            remote_rows.append(
                {
                    "remote": remote_name,
                    "endpoint": "",
                    "type": "unsupported",
                    "org": "",
                    "repo": "",
                    "url_kind": "unsupported",
                    "url": safe_remote_url_label(remote_url),
                    "token_id": "",
                    "token_value": "",
                    "result": "ignored",
                }
            )
            continue

        touched_servers.add(server_info["endpoint"])
        credentials = remote_credentials(remote_url)
        if credentials is None:
            plain_urls += 1
            remote_rows.append(
                {
                    "remote": remote_name,
                    "endpoint": server_info["endpoint"],
                    "type": server_info["type"],
                    "org": server_info["org"],
                    "repo": server_info["repo"],
                    "url_kind": "plain",
                    "url": safe_remote_url_label(remote_url),
                    "token_id": "",
                    "token_value": "",
                    "result": "no_credentials",
                }
            )
            continue

        auth_urls += 1
        user_name, token_value = credentials
        row_index = len(remote_rows)
        remote_rows.append(
            {
                "remote": remote_name,
                "endpoint": server_info["endpoint"],
                "type": server_info["type"],
                "org": server_info["org"],
                "repo": server_info["repo"],
                "url_kind": "auth",
                "url": safe_remote_url_label(remote_url),
                "token_id": remote_name,
                "user": user_name,
                "token_value": token_value,
                "result": "found",
            }
        )
        found_credentials.append((row_index, server_info, remote_name, user_name, token_value))

    return {
        "repo_root": repo_root,
        "remotes": len(remote_names),
        "urls": len(remote_entries),
        "scanned": auth_urls + plain_urls,
        "auth_urls": auth_urls,
        "plain_urls": plain_urls,
        "unsupported_urls": unsupported_urls,
        "found": len(found_credentials),
        "added": 0,
        "updated": 0,
        "servers": len(touched_servers),
        "remote_rows": remote_rows,
        "found_credentials": found_credentials,
        "changed": False,
    }


def sync_tokens_from_repo(config: WorkspaceConfig, repo_path: Path) -> dict:
    report = scan_repo_remotes(config, repo_path)
    found_credentials = report.get("found_credentials", [])
    if not found_credentials:
        return report

    token_data = load_token_store(config)
    added = 0
    updated = 0
    for row_index, server_info, remote_name, user_name, token_value in found_credentials:
        result = register_token(token_data, server_info, remote_name, user_name, token_value)
        if result == "added":
            added += 1
        elif result == "updated":
            updated += 1
        report["remote_rows"][row_index]["result"] = result

    changed = added > 0 or updated > 0
    if changed or not config.token_path.exists():
        write_token_store(config, token_data)

    report["added"] = added
    report["updated"] = updated
    report["changed"] = changed
    return report


def sync_token_from_remote(config: WorkspaceConfig, repo_path: Path, remote_name: str, dry_run: bool = False) -> dict:
    repo_root = git_repo_root(repo_path)
    if repo_root is None:
        raise SystemExit(f"Missing git repo at path: {repo_path}")

    remote_url = git_capture(repo_root, ["remote", "get-url", remote_name])
    if not remote_url:
        raise SystemExit(f"Missing git remote: {remote_name}")
    server_info = server_info_from_url(config, remote_url)
    if server_info is None:
        raise SystemExit(f"Only http/https remote URLs are supported: {remote_url}")
    credentials = remote_credentials(remote_url)
    if credentials is None:
        raise SystemExit(f"Remote '{remote_name}' has no credentials.")

    user_name, token_value = credentials
    result = "dry-run"
    if not dry_run:
        token_data = load_token_store(config)
        result = register_token(
            token_data,
            server_info,
            remote_name,
            user_name,
            token_value,
            hydrate=False,
            reset_api_metadata=True,
        )
        write_token_store(config, token_data)

    return {
        "repo_root": str(repo_root),
        "remote": remote_name,
        "server": server_info["endpoint"],
        "user": user_name,
        "org": server_info.get("org", ""),
        "repo": server_info.get("repo", ""),
        "status": result,
    }


def hydrate_token_from_api(config: WorkspaceConfig, remote_name: str, endpoint: str | None, dry_run: bool = False) -> dict:
    token_data = load_token_store(config, write_normalized=not dry_run)
    token_record = resolve_token_from_store(token_data, endpoint, remote_name, None)
    token_value = str(token_record.get("value", ""))
    if not token_value:
        raise SystemExit(f"Token value is empty in tokens.json: {token_server_endpoint(token_record)}:{remote_name}")

    authz = load_gitea_authz(
        token_server_endpoint(token_record),
        remote_name,
        token_value,
        token_record.get("expires_at", ""),
        token_record.get("org", ""),
        token_record.get("repo", ""),
    )
    changed = apply_authz_to_token(token_record, authz)
    if changed and not dry_run:
        write_token_store(config, token_data)

    return {
        "remote": remote_name,
        "server": token_server_endpoint(token_record),
        "user": str(token_record.get("user", "")),
        "valid": str(token_record.get("valid", "")),
        "status": "dry-run" if dry_run else ("updated" if changed else "unchanged"),
    }


def count_server_tokens(server_entry: dict) -> int:
    return len(server_entry.get("tokens", []))


def mask_secret(secret_value: str) -> str:
    if len(secret_value) <= 4:
        return "*" * len(secret_value)
    return f"{secret_value[:2]}{'*' * (len(secret_value) - 4)}{secret_value[-2:]}"


def store_servers_from_tokens(token_data: dict) -> dict[str, dict]:
    servers: dict[str, dict] = {}
    for token_record in token_data.get("tokens", []):
        server_record = token_record.get("server", {})
        if not isinstance(server_record, dict):
            continue
        endpoint = str(server_record.get("endpoint", ""))
        if not endpoint:
            continue

        server_entry = servers.setdefault(
            endpoint,
            {
                "type": server_record.get("type", "unknown"),
                "scheme": server_record.get("scheme", ""),
                "host": server_record.get("host", ""),
                "port": server_record.get("port"),
                "tokens": [],
            },
        )
        for field_name in ["type", "scheme", "host", "port"]:
            if server_entry.get(field_name) in {"", None, "unknown"} and server_record.get(field_name) not in {"", None}:
                server_entry[field_name] = server_record.get(field_name)

        token_id = token_record_id(token_record)
        if not token_id:
            continue
        server_entry["tokens"].append(
            {
                "endpoint": endpoint,
                "token_id": token_id,
                "token_value": str(token_record.get("value", "")),
                "user": str(token_record.get("user", "")),
                "valid": str(token_record.get("valid", "")),
                "scope_mask": permission_map_to_mask(token_record.get("scope"), SCOPE_FIELDS, "rw?!-"),
                "expires_at": str(token_record.get("expires_at", "")),
                "remote": token_id,
                "owner": str(token_record.get("org", "")),
                "repo_name": str(token_record.get("repo", "")),
                "org_mask": permission_map_to_mask(token_record.get("org_perm"), ORG_PERMISSION_FIELDS, "+?!-"),
                "repo_mask": permission_map_to_mask(token_record.get("repo_perm"), REPO_PERMISSION_FIELDS, "+?!-"),
            }
        )
    return servers


def token_store_servers(token_data: dict) -> list[tuple[str, dict]]:
    return sorted(store_servers_from_tokens(token_data).items())


def token_pairs_from_entry(server_entry: dict) -> set[tuple[str, str, str, str, str]]:
    token_pairs: set[tuple[str, str, str, str, str]] = set()
    for row in server_entry.get("tokens", []):
        token_id = row.get("token_id", "")
        token_value = row.get("token_value", "")
        if token_id and token_value:
            token_pairs.add(
                (
                    str(row.get("remote", "")),
                    str(row.get("user", "")),
                    str(token_value),
                    str(row.get("org", row.get("owner", ""))),
                    str(row.get("repo", row.get("repo_name", ""))),
                )
            )
    return token_pairs


def count_users(users_data: dict) -> int:
    return len(users_data)


def collect_repo_server_entries(config: WorkspaceConfig, repo_path: Path) -> tuple[Path | None, dict[str, dict]]:
    repo_root = git_repo_root(repo_path)
    if repo_root is None:
        return None, {}

    repo_servers: dict[str, dict] = {}
    for remote_name, remote_url in remote_name_urls(repo_root):
        server_info = server_info_from_url(config, remote_url)
        if server_info is None:
            continue

        endpoint = server_info["endpoint"]
        entry = repo_servers.setdefault(
            endpoint,
            {
                "type": server_info["type"],
                "scheme": server_info["scheme"],
                "host": server_info["host"],
                "port": server_info["port"],
                "remotes": set(),
                "urls": 0,
                "auth_urls": 0,
                "plain_urls": 0,
                "tokens": [],
                "plain_remotes": [],
            },
        )
        entry["remotes"].add(remote_name)
        entry["urls"] += 1
        for field_name in ["type", "scheme", "host", "port"]:
            if field_name not in entry or entry[field_name] in {"", None, "unknown"}:
                entry[field_name] = server_info[field_name]

        credentials = remote_credentials(remote_url)
        if credentials is None:
            entry["plain_urls"] += 1
            entry["plain_remotes"].append(
                {
                    "remote": remote_name,
                    "org": server_info["org"],
                    "repo": server_info["repo"],
                }
            )
            continue

        entry["auth_urls"] += 1
        user_name, token_value = credentials
        entry["tokens"].append(
            {
                "remote": remote_name,
                "org": server_info["org"],
                "repo": server_info["repo"],
                "token_id": remote_name,
                "user": user_name,
                "token_value": token_value,
            }
        )

    return repo_root, repo_servers


def compare_server_entries(repo_entry: dict | None, store_entry: dict | None) -> tuple[str, int, int]:
    if repo_entry is None and store_entry is None:
        return "missing", 0, 0
    if repo_entry is not None and store_entry is None:
        return "repo_only", count_server_tokens(repo_entry), 0
    if repo_entry is None and store_entry is not None:
        return "store_only", 0, count_server_tokens(store_entry)

    repo_pairs = token_pairs_from_entry(repo_entry)
    store_pairs = token_pairs_from_entry(store_entry)
    repo_only_pairs = repo_pairs - store_pairs
    store_only_pairs = store_pairs - repo_pairs
    if not repo_only_pairs and not store_only_pairs:
        return "in_sync", 0, 0
    if repo_only_pairs and not store_only_pairs:
        return "repo_ahead", len(repo_only_pairs), 0
    if store_only_pairs and not repo_only_pairs:
        return "store_ahead", 0, len(store_only_pairs)
    return "diverged", len(repo_only_pairs), len(store_only_pairs)


def repo_result_for_entry(repo_entry: dict | None) -> tuple[str, str]:
    if repo_entry is None:
        return "missing", "missing"
    if repo_entry.get("auth_urls", 0):
        return "auth", "present"
    if repo_entry.get("plain_urls", 0):
        return "plain", "no_credentials"
    return "empty", "no_credentials"


def print_token_context(
    repo_root: Path | None,
    config: WorkspaceConfig,
    repo_servers: dict[str, dict],
    store_servers: dict[str, dict],
    endpoint_count: int,
) -> None:
    print("context")
    print("item\tvalue")
    print(f"repo_root\t{repo_root or ''}")
    print(f"token_path\t{config.token_path}")
    print(f"repo_endpoints\t{len(repo_servers)}")
    print(f"tokens_endpoints\t{len(store_servers)}")
    print(f"tokens\t{endpoint_count}")


def print_status_counts(status_counts: dict[str, int]) -> None:
    print("status")
    print("item\tvalue")
    for status_name in ["in_sync", "repo_only", "store_only", "repo_ahead", "store_ahead", "diverged"]:
        print(f"{status_name}\t{status_counts[status_name]}")


TOKEN_REF_WIDTH = 9

TOKEN_COLUMNS = [
    ("item", "item", 4),
    ("server", "server", 6),
    ("proto", "proto", 5),
    ("host", "host", 18),
    ("owner", "org", 9),
    ("repo_name", "repo", 11),
    ("user", "user", 4),
    ("remote", "remote", 6),
    ("token_ref", "token_ref", TOKEN_REF_WIDTH),
    ("token", "token", 12),
    ("valid", "valid", 19),
    ("scope_mask", "scope", 9),
    ("org_mask", "org", 6),
    ("repo_mask", "repo", 6),
]

TOKEN_LIST_COLUMNS = [
    ("item", "item", 4),
    ("source", "source", 6),
    ("kind", "kind", 5),
    ("server", "server", 6),
    ("proto", "proto", 5),
    ("host", "host", 18),
    ("owner", "org", 9),
    ("repo_name", "repo", 11),
    ("user", "user", 4),
    ("remote", "remote", 6),
    ("token", "token", 12),
    ("valid", "valid", 19),
    ("scope_mask", "scope", 9),
    ("org_mask", "org", 6),
    ("repo_mask", "repo", 6),
]

TOKEN_SCAN_COLUMNS = [
    ("item", "item", 4),
    ("remote", "remote", 6),
    ("kind", "kind", 11),
    ("server", "server", 6),
    ("proto", "proto", 5),
    ("host", "host", 18),
    ("owner", "org", 9),
    ("repo_name", "repo", 11),
    ("user", "user", 4),
    ("token", "token", 12),
    ("result", "result", 14),
    ("url", "url", 38),
]

TOKEN_SEPARATOR_OVERRIDES = {
    "scope_mask": "aAimnopru",
    "org_mask": "oawrc-",
    "repo_mask": "oawr--",
}


def fit_cell(value: str, width: int) -> str:
    if len(value) <= width:
        return value.ljust(width)
    if width <= 3:
        return value[:width]
    return f"{value[:width - 3]}..."


def print_fixed_table(title: str, columns: list[tuple[str, str, int]], rows: list[dict[str, str]]) -> None:
    print(title)
    print("  ".join(fit_cell(label, width) for _, label, width in columns).rstrip())
    print("  ".join(fit_cell(TOKEN_SEPARATOR_OVERRIDES.get(name, "-" * width), width) for name, _, width in columns).rstrip())
    for row in rows:
        print("  ".join(fit_cell(row.get(name, ""), width) for name, _, width in columns).rstrip())


def short_secret(secret_value: str) -> str:
    if len(secret_value) <= 12:
        return mask_secret(secret_value)
    return f"{secret_value[:5]}...{secret_value[-4:]}"


def format_token_ref(token_name: str, marker: str) -> str:
    name_width = TOKEN_REF_WIDTH - 2
    return f"{fit_cell(token_name, name_width)}{marker:>2}"


def users_label(server_entry: dict | None) -> str:
    if server_entry is None:
        return ""
    return ",".join(sorted(str(user_name) for user_name in server_entry.get("users", {})))


def endpoint_column_values(endpoint: str, repo_entry: dict | None, store_entry: dict | None) -> dict[str, str]:
    values = {
        "server": "",
        "scheme": "",
        "host": "",
        "port": "",
    }
    for server_entry in [repo_entry, store_entry]:
        if server_entry is None:
            continue
        values["server"] = values["server"] or str(server_entry.get("type", ""))
        values["scheme"] = values["scheme"] or str(server_entry.get("scheme", ""))
        values["host"] = values["host"] or str(server_entry.get("host", ""))
        port_value = server_entry.get("port", "")
        values["port"] = values["port"] or ("" if port_value is None else str(port_value))

    split_endpoint = urlsplit(endpoint)
    values["scheme"] = values["scheme"] or split_endpoint.scheme
    values["host"] = values["host"] or (split_endpoint.hostname or "")
    if not values["port"] and split_endpoint.port is not None:
        values["port"] = str(split_endpoint.port)
    if values["host"] and values["port"]:
        values["host"] = f"{values['host']}:{values['port']}"
    return values


def token_store_rows(store_servers: dict[str, dict]) -> list[dict[str, str]]:
    rows: list[dict[str, str]] = []
    for endpoint, server_entry in sorted(store_servers.items()):
        for row in server_entry.get("tokens", []):
            token_value = row.get("token_value", "")
            token_id = row.get("token_id", "")
            if token_id:
                rows.append(
                    {
                        "endpoint": endpoint,
                        "user": str(row.get("user", "")),
                        "token_id": str(token_id),
                        "remote": str(row.get("remote", "")),
                        "owner": str(row.get("owner", "")),
                        "repo_name": str(row.get("repo_name", "")),
                        "token_value": str(token_value),
                        "expires_at": str(row.get("expires_at", "")),
                        "valid": str(row.get("valid", "")),
                        "scope_mask": str(row.get("scope_mask", "")),
                        "org_mask": str(row.get("org_mask", "")),
                        "repo_mask": str(row.get("repo_mask", "")),
                    }
                )
    return rows


def repo_token_rows(
    endpoint_names: list[str],
    repo_servers: dict[str, dict],
    scan_rows: list[dict[str, str]] | None,
) -> list[dict[str, str]]:
    if scan_rows is not None:
        return [
            {
                "endpoint": row["endpoint"],
                "remote": row["remote"],
                "owner": row.get("org", ""),
                "repo_name": row.get("repo", ""),
                "token_id": row.get("token_id", ""),
                "user": row.get("user", ""),
                "token_value": row.get("token_value", ""),
            }
            for row in scan_rows
            if row.get("endpoint") and row.get("url_kind") == "auth" and row.get("token_value")
        ]

    rows: list[dict[str, str]] = []
    for endpoint in endpoint_names:
        repo_entry = repo_servers.get(endpoint)
        if repo_entry is None:
            continue
        for row in repo_entry.get("tokens", []):
            rows.append(
                {
                    "endpoint": endpoint,
                    "remote": row.get("remote", ""),
                    "owner": row.get("org", ""),
                    "repo_name": row.get("repo", ""),
                    "token_id": row.get("token_id", ""),
                    "user": row.get("user", ""),
                    "token_value": row.get("token_value", ""),
                }
            )
    return rows


def list_store_rows(store_servers: dict[str, dict], server_filter: str | None = None) -> list[dict[str, str]]:
    rows: list[dict[str, str]] = []
    endpoints = sorted(store_servers)
    for endpoint in endpoints:
        if server_filter and endpoint != server_filter:
            continue
        server_entry = store_servers[endpoint]
        endpoint_values = endpoint_column_values(endpoint, None, server_entry)
        for row in sorted(server_entry.get("tokens", []), key=lambda item: item.get("token_id", "")):
            token_value = str(row.get("token_value", ""))
            rows.append(
                {
                    "item": str(len(rows) + 1),
                    "source": "store",
                    "kind": "auth" if token_value else "empty",
                    "server": endpoint_values["server"],
                    "proto": endpoint_values["scheme"],
                    "host": endpoint_values["host"],
                    "owner": str(row.get("owner", "")),
                    "repo_name": str(row.get("repo_name", "")),
                    "user": str(row.get("user", "")),
                    "remote": str(row.get("remote", "")),
                    "token": short_secret(token_value) if token_value else "",
                    "valid": str(row.get("valid", "")),
                    "scope_mask": str(row.get("scope_mask", "")),
                    "org_mask": str(row.get("org_mask", "")),
                    "repo_mask": str(row.get("repo_mask", "")),
                }
            )
    return rows


def list_remote_rows(repo_servers: dict[str, dict], server_filter: str | None = None) -> list[dict[str, str]]:
    rows: list[dict[str, str]] = []
    endpoints = sorted(repo_servers)
    for endpoint in endpoints:
        if server_filter and endpoint != server_filter:
            continue
        server_entry = repo_servers[endpoint]
        endpoint_values = endpoint_column_values(endpoint, server_entry, None)
        remote_rows = []
        for row in server_entry.get("tokens", []):
            remote_rows.append(
                {
                    "kind": "auth",
                    "remote": row.get("remote", ""),
                    "owner": row.get("org", ""),
                    "repo_name": row.get("repo", ""),
                    "user": row.get("user", ""),
                    "token_value": row.get("token_value", ""),
                }
            )
        for row in server_entry.get("plain_remotes", []):
            remote_rows.append(
                {
                    "kind": "plain",
                    "remote": row.get("remote", ""),
                    "owner": row.get("org", ""),
                    "repo_name": row.get("repo", ""),
                    "user": "",
                    "token_value": "",
                }
            )

        for row in sorted(remote_rows, key=lambda item: (item.get("remote", ""), item.get("kind", ""))):
            token_value = str(row.get("token_value", ""))
            rows.append(
                {
                    "item": str(len(rows) + 1),
                    "source": "remote",
                    "kind": str(row.get("kind", "")),
                    "server": endpoint_values["server"],
                    "proto": endpoint_values["scheme"],
                    "host": endpoint_values["host"],
                    "owner": str(row.get("owner", "")),
                    "repo_name": str(row.get("repo_name", "")),
                    "user": str(row.get("user", "")),
                    "remote": str(row.get("remote", "")),
                    "token": short_secret(token_value) if token_value else "",
                    "valid": "?",
                    "scope_mask": "?????????",
                    "org_mask": "?????",
                    "repo_mask": "????",
                }
            )
    return rows


def scan_remote_rows(report: dict, server_filter: str | None = None) -> list[dict[str, str]]:
    rows: list[dict[str, str]] = []
    for row in report.get("remote_rows", []):
        endpoint = str(row.get("endpoint", ""))
        if server_filter and endpoint != server_filter:
            continue
        endpoint_values = endpoint_column_values(endpoint, None, None) if endpoint else {}
        token_value = str(row.get("token_value", ""))
        rows.append(
            {
                "item": str(len(rows) + 1),
                "remote": str(row.get("remote", "")),
                "kind": str(row.get("url_kind", "")),
                "server": str(row.get("type", "")),
                "proto": endpoint_values.get("scheme", ""),
                "host": endpoint_values.get("host", ""),
                "owner": str(row.get("org", "")),
                "repo_name": str(row.get("repo", "")),
                "user": str(row.get("user", "")),
                "token": short_secret(token_value) if token_value else "",
                "result": str(row.get("result", "")),
                "url": str(row.get("url", "")),
            }
        )
    return rows


def fallback_project_by_endpoint(
    endpoint_names: list[str],
    repo_servers: dict[str, dict],
    scan_rows: list[dict[str, str]] | None,
) -> dict[str, dict[str, str]]:
    fallback: dict[str, dict[str, str]] = {}
    for row in scan_rows or []:
        endpoint = row.get("endpoint", "")
        if endpoint and endpoint not in fallback:
            fallback[endpoint] = {
                "remote": row.get("remote", ""),
                "owner": row.get("org", ""),
                "repo_name": row.get("repo", ""),
            }

    for endpoint in endpoint_names:
        if endpoint in fallback:
            continue
        repo_entry = repo_servers.get(endpoint)
        if repo_entry is None:
            continue
        for row in repo_entry.get("tokens", []) + repo_entry.get("plain_remotes", []):
            fallback[endpoint] = {
                "remote": row.get("remote", ""),
                "owner": row.get("org", ""),
                "repo_name": row.get("repo", ""),
            }
            break
    return fallback


def api_get_json(api_url: str, token_value: str, user_name: str = "", basic: bool = False) -> object:
    headers = {"Accept": "application/json"}
    if basic:
        raw_credentials = f"{user_name}:{token_value}".encode("utf-8")
        headers["Authorization"] = "Basic " + base64.b64encode(raw_credentials).decode("ascii")
    else:
        headers["Authorization"] = f"token {token_value}"
    request = urlrequest.Request(api_url, headers=headers)
    with urlrequest.urlopen(request, timeout=5) as response:
        return json.loads(response.read().decode("utf-8"))


def scope_mask(scopes: list[str]) -> str:
    scope_set = {scope.strip().lower() for scope in scopes if scope.strip()}
    if "all" in scope_set:
        return "w" * len(SCOPE_FIELDS)

    values = []
    for _, category in SCOPE_FIELDS:
        category_names = SCOPE_CATEGORY_ALIASES.get(category, [category])
        if any(f"write:{category_name}" in scope_set for category_name in category_names):
            values.append("w")
        elif any(f"read:{category_name}" in scope_set for category_name in category_names):
            values.append("r")
        else:
            values.append("-")
    return "".join(values)


def org_permission_mask(org_permissions: dict) -> str:
    return "".join(
        [
            "+" if org_permissions.get("is_owner") else "-",
            "+" if org_permissions.get("is_admin") else "-",
            "+" if org_permissions.get("can_write") else "-",
            "+" if org_permissions.get("can_read") else "-",
            "+" if org_permissions.get("can_create_repository") else "-",
        ]
    )


def repo_permission_mask(repo_permissions: dict) -> str:
    permission = str(repo_permissions.get("permission", "none")).lower()
    levels = {
        "owner": "++++",
        "admin": "-+++",
        "write": "--++",
        "read": "---+",
    }
    return levels.get(permission, "----")


def load_gitea_authz(
    endpoint: str,
    token_id: str,
    token_value: str,
    expires_at: str,
    owner: str,
    repo_name: str,
) -> dict:
    api_base = endpoint.rstrip("/") + "/api/v1"
    result = {
        "ok": False,
        "valid": "!",
        "scope_mask": "!!!!!!!!!",
        "org_mask": "!!!!!",
        "repo_mask": "!!!!",
        "user": "",
    }

    try:
        user_info = api_get_json(f"{api_base}/user", token_value)
        user_name = str(user_info.get("login", "")) if isinstance(user_info, dict) else ""
        result["user"] = user_name
        result["valid"] = valid_label(True, expires_at)

        if user_name:
            try:
                tokens_url = f"{api_base}/users/{quote(user_name, safe='')}/tokens"
                token_list = api_get_json(tokens_url, token_value, user_name=user_name, basic=True)
                token_last8 = token_value[-8:]
                token_entry = next(
                    (
                        entry
                        for entry in token_list
                        if isinstance(entry, dict) and entry.get("token_last_eight") == token_last8
                    ),
                    None,
                )
                scopes = token_entry.get("scopes", []) if isinstance(token_entry, dict) else []
                if not isinstance(scopes, list):
                    scopes = []
                result["scope_mask"] = scope_mask([str(scope) for scope in scopes])
            except (OSError, ValueError, urlerror.URLError):
                result["scope_mask"] = "?????????"

        if user_name and owner:
            org_url = f"{api_base}/users/{quote(user_name, safe='')}/orgs/{quote(owner, safe='')}/permissions"
            org_permissions = api_get_json(org_url, token_value)
            if isinstance(org_permissions, dict):
                result["org_mask"] = org_permission_mask(org_permissions)

        if user_name and owner and repo_name:
            repo_url = (
                f"{api_base}/repos/{quote(owner, safe='')}/{quote(repo_name, safe='')}"
                f"/collaborators/{quote(user_name, safe='')}/permission"
            )
            repo_permissions = api_get_json(repo_url, token_value)
            if isinstance(repo_permissions, dict):
                result["repo_mask"] = repo_permission_mask(repo_permissions)

        result["ok"] = True
        return result
    except urlerror.HTTPError as error:
        result["valid"] = "invalid" if error.code in {401, 403} else "!"
        return result
    except (OSError, ValueError, urlerror.URLError):
        return result


def print_token_item_rows(
    endpoint_names: list[str],
    repo_servers: dict[str, dict],
    store_servers: dict[str, dict],
    scan_rows: list[dict[str, str]] | None = None,
) -> dict[str, int]:
    status_names = ["in_sync", "repo_only", "store_only", "repo_ahead", "store_ahead", "diverged"]
    status_counts = {status_name: 0 for status_name in status_names}
    for endpoint in endpoint_names:
        repo_entry = repo_servers.get(endpoint)
        store_entry = store_servers.get(endpoint)
        status_name, _, _ = compare_server_entries(repo_entry, store_entry)
        if status_name in status_counts:
            status_counts[status_name] += 1

    store_rows = token_store_rows(store_servers)
    store_by_remote: dict[tuple[str, str], dict[str, str]] = {}
    for row in store_rows:
        if row["remote"]:
            store_by_remote.setdefault((row["endpoint"], row["remote"]), row)

    endpoint_values_by_name = {
        endpoint: endpoint_column_values(endpoint, repo_servers.get(endpoint), store_servers.get(endpoint))
        for endpoint in endpoint_names
    }
    project_fallback = fallback_project_by_endpoint(endpoint_names, repo_servers, scan_rows)
    used_store_keys: set[tuple[str, str, str]] = set()
    token_rows: list[dict[str, str]] = []

    def append_token_row(
        endpoint: str,
        user_name: str,
        remote_name: str,
        token_id: str,
        token_value: str,
        marker: str,
        owner: str,
        repo_name: str,
        store_row: dict | None = None,
    ) -> None:
        endpoint_values = endpoint_values_by_name.get(endpoint, endpoint_column_values(endpoint, None, None))
        valid_value = store_row.get("valid", "?") if store_row else "?"
        scope_value = store_row.get("scope_mask", "?????????") if store_row else "?????????"
        org_value = store_row.get("org_mask", "?????") if store_row else "?????"
        repo_value = store_row.get("repo_mask", "????") if store_row else "????"
        token_rows.append(
            {
                "item": str(len(token_rows) + 1),
                "server": endpoint_values["server"],
                "proto": endpoint_values["scheme"],
                "host": endpoint_values["host"],
                "owner": owner,
                "repo_name": repo_name,
                "user": user_name,
                "remote": remote_name,
                "token_ref": format_token_ref(token_id, marker),
                "token": short_secret(token_value),
                "valid": valid_value or "?",
                "scope_mask": scope_value or "?????????",
                "org_mask": org_value or "?????",
                "repo_mask": repo_value or "????",
            }
        )

    for row in repo_token_rows(endpoint_names, repo_servers, scan_rows):
        endpoint = row["endpoint"]
        token_value = row["token_value"]
        store_row = store_by_remote.get((endpoint, row["remote"]))
        token_id = row["token_id"]
        expires_at = store_row.get("expires_at", "") if store_row else ""
        marker = "R"
        synced_store_row = None
        if store_row:
            repo_fields_match = (
                store_row["token_id"] == row["token_id"]
                and store_row["user"] == row.get("user", "")
                and store_row["token_value"] == token_value
                and store_row["owner"] == row["owner"]
                and store_row["repo_name"] == row["repo_name"]
            )
            if repo_fields_match:
                used_store_keys.add((endpoint, store_row["token_id"], store_row["remote"]))
                synced_store_row = store_row
                marker = "!" if store_row.get("valid") in {"!", "invalid"} else "*"
        append_token_row(
            endpoint,
            synced_store_row.get("user", "") if synced_store_row else row.get("user", ""),
            row["remote"],
            token_id,
            token_value,
            marker,
            row["owner"],
            row["repo_name"],
            synced_store_row,
        )

    for row in store_rows:
        store_key = (row["endpoint"], row["token_id"], row["remote"])
        if store_key in used_store_keys:
            continue
        fallback = project_fallback.get(row["endpoint"], {})
        append_token_row(
            row["endpoint"],
            row["user"],
            row["remote"],
            row["token_id"],
            row["token_value"],
            "S",
            row["owner"] or fallback.get("owner", ""),
            row["repo_name"] or fallback.get("repo_name", ""),
        )

    print_fixed_table("tokens", TOKEN_COLUMNS, token_rows)
    return status_counts


def resolve_token_from_store(
    token_data: dict,
    endpoint: str | None,
    remote_id: str | None,
    user_name: str | None,
) -> dict:
    candidates = []
    for token_record in token_data.get("tokens", []):
        token_endpoint = token_server_endpoint(token_record)
        if endpoint and token_endpoint != endpoint:
            continue
        if remote_id and token_record_id(token_record) != remote_id:
            continue
        if user_name and token_record.get("user") != user_name:
            continue
        candidates.append(token_record)

    if not candidates:
        raise SystemExit("No matching token in tokens.json.")
    if len(candidates) > 1:
        available = ", ".join(
            sorted(
                f"{token_server_endpoint(token_record)}:{token_record_id(token_record)}"
                for token_record in candidates
            )
        )
        raise SystemExit(f"Multiple matching tokens. Pass --server/--token-name or --remote. Available: {available}")
    return candidates[0]


def sanitize_remote_url(remote_url: str) -> tuple[str, bool]:
    split_url = urlsplit(remote_url)
    if split_url.scheme not in {"http", "https"} or split_url.hostname is None:
        raise SystemExit(f"Only http/https remote URLs are supported: {remote_url}")

    _, port_value = endpoint_from_split(split_url)
    default_port = default_port_for_scheme(split_url.scheme)
    netloc = split_url.hostname
    if port_value is not None and port_value != default_port:
        netloc = f"{netloc}:{port_value}"

    had_credentials = split_url.username is not None or split_url.password is not None
    sanitized_url = urlunsplit((split_url.scheme, netloc, split_url.path, split_url.query, split_url.fragment))
    return sanitized_url, had_credentials


def credentialed_remote_url(remote_url: str, user_name: str, token_value: str) -> str:
    sanitized_url, _ = sanitize_remote_url(remote_url)
    split_url = urlsplit(sanitized_url)
    _, port_value = endpoint_from_split(split_url)
    default_port = default_port_for_scheme(split_url.scheme)
    host_part = split_url.hostname or ""
    if port_value is not None and port_value != default_port:
        host_part = f"{host_part}:{port_value}"

    netloc = f"{quote(user_name, safe='')}:{quote(token_value, safe='')}@{host_part}"
    return urlunsplit((split_url.scheme, netloc, split_url.path, split_url.query, split_url.fragment))


def remote_url_from_token_record(token_record: dict) -> str:
    endpoint = token_server_endpoint(token_record).rstrip("/")
    owner = str(token_record.get("org", "")).strip("/")
    repo_name = str(token_record.get("repo", "")).strip("/")
    if not endpoint or not owner or not repo_name:
        return ""
    if not repo_name.endswith(".git"):
        repo_name = f"{repo_name}.git"
    return f"{endpoint}/{quote(owner, safe='')}/{quote(repo_name, safe='')}"


def remote_urls_point_to_same_repo(left_url: str, right_url: str) -> bool:
    try:
        left_sanitized, _ = sanitize_remote_url(left_url)
        right_sanitized, _ = sanitize_remote_url(right_url)
    except SystemExit:
        return False
    return left_sanitized.rstrip("/") == right_sanitized.rstrip("/")


def find_origin_rename_candidate(
    config: WorkspaceConfig,
    repo_root: Path,
    target_remote: str,
    target_url: str,
) -> tuple[str, str] | None:
    origin_remote = config.git.origin_remote
    if not origin_remote or origin_remote == target_remote:
        return None
    if git_capture(repo_root, ["remote", "get-url", target_remote]) is not None:
        return None

    origin_url = git_capture(repo_root, ["remote", "get-url", origin_remote])
    if not origin_url:
        return None
    if not remote_urls_point_to_same_repo(origin_url, target_url):
        return None
    return origin_remote, origin_url


def resolve_repo_argument(repo_arg: str | None) -> Path:
    return Path(repo_arg).expanduser().resolve() if repo_arg else Path.cwd().resolve()


def resolve_launcher_repo_argument(repo_arg: str | None) -> Path:
    return Path(repo_arg).expanduser().resolve() if repo_arg else Path(__file__).resolve().parent


def safe_remote_url_label(remote_url: str | None) -> str:
    if not remote_url:
        return ""
    split_url = urlsplit(remote_url)
    if split_url.scheme in {"http", "https"} and split_url.hostname is not None:
        return sanitize_remote_url(remote_url)[0]
    return remote_url


def run_tokens_scan(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    repo_path = resolve_launcher_repo_argument(args.repo)
    report = scan_repo_remotes(config, repo_path)
    print_fixed_table("remotes", TOKEN_SCAN_COLUMNS, scan_remote_rows(report, args.server))


def run_tokens_compare(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    repo_path = resolve_launcher_repo_argument(args.repo)
    token_data = load_token_store(config, write_normalized=False)
    store_servers = store_servers_from_tokens(token_data)
    report = scan_repo_remotes(config, repo_path)
    repo_root, repo_servers = collect_repo_server_entries(config, repo_path)
    endpoint_names = sorted(set(repo_servers) | set(store_servers))
    print_token_item_rows(endpoint_names, repo_servers, store_servers, report["remote_rows"])


def run_tokens_list(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    rows: list[dict[str, str]] = []
    if args.target in {"store", "both"}:
        token_data = load_token_store(config, write_normalized=False)
        store_servers = store_servers_from_tokens(token_data)
        rows.extend(list_store_rows(store_servers, args.server))

    if args.target in {"remote", "both"}:
        repo_root, repo_servers = collect_repo_server_entries(config, resolve_launcher_repo_argument(args.repo))
        if repo_root is None:
            raise SystemExit(f"Missing git repo at path: {resolve_launcher_repo_argument(args.repo)}")
        rows.extend(list_remote_rows(repo_servers, args.server))

    for index, row in enumerate(rows, start=1):
        row["item"] = str(index)
    print_fixed_table("tokens", TOKEN_LIST_COLUMNS, rows)


def run_tokens_read(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    token_data = load_token_store(config, write_normalized=False)
    servers = token_store_servers(token_data)
    if args.server:
        server_entry = dict(store_servers_from_tokens(token_data)).get(args.server)
        if server_entry is None:
            raise SystemExit(f"Missing server endpoint in token store: {args.server}")
        servers = [(args.server, server_entry)]

    if not servers:
        print(f"token_path\t{config.token_path}")
        print("servers\t0")
        return

    print(f"token_path\t{config.token_path}")
    for index, (endpoint, server_entry) in enumerate(servers):
        if index:
            print()
        print(f"endpoint\t{endpoint}")
        print(f"type\t{server_entry.get('type', 'unknown')}")
        print(f"scheme\t{server_entry.get('scheme', '')}")
        print(f"host\t{server_entry.get('host', '')}")
        print(f"port\t{server_entry.get('port', '')}")
        print(f"tokens\t{count_server_tokens(server_entry)}")
        for row in sorted(server_entry.get("tokens", []), key=lambda item: item.get("token_id", "")):
            token_value = row.get("token_value", "")
            secret_value = token_value if args.show_secrets else mask_secret(token_value)
            print(f"id\t{row.get('token_id', '')}\t{secret_value}")
            for field_name in ["user", "valid", "scope_mask", "expires_at"]:
                field_value = row.get(field_name, "")
                if field_value:
                    print(f"{field_name}\t{row.get('token_id', '')}\t{field_value}")
            if row.get("remote"):
                print(
                    "remote\t"
                    f"{row.get('token_id', '')}\t{row.get('remote', '')}\t"
                    f"{row.get('owner', '')}\t{row.get('repo_name', '')}"
                )
            for field_name in ["org_mask", "repo_mask"]:
                field_value = row.get(field_name, "")
                if field_value:
                    print(f"{field_name}\t{row.get('token_id', '')}\t{field_value}")


def run_tokens_stats(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    repo_root, repo_servers = collect_repo_server_entries(config, resolve_launcher_repo_argument(args.repo))
    token_data = load_token_store(config, write_normalized=False)
    store_servers = store_servers_from_tokens(token_data)
    endpoint_names = sorted(set(repo_servers) | set(store_servers))
    if args.server:
        if args.server not in set(endpoint_names):
            raise SystemExit(f"Missing server endpoint in repo/store: {args.server}")
        endpoint_names = [args.server]

    print_token_context(repo_root, config, repo_servers, store_servers, len(endpoint_names))
    print()
    status_counts = print_token_item_rows(endpoint_names, repo_servers, store_servers)
    print()
    print_status_counts(status_counts)


def print_token_action_result(result: dict[str, str]) -> None:
    for field_name in ["repo_root", "token_path", "remote", "server", "user", "org", "repo", "valid", "url", "removed", "status"]:
        field_value = result.get(field_name)
        if field_value is not None:
            print(f"{field_name}\t{field_value}")


def remove_token_from_store(
    config: WorkspaceConfig,
    remote_id: str,
    endpoint: str | None = None,
    dry_run: bool = False,
) -> dict[str, str]:
    token_data = load_token_store(config, write_normalized=False)
    matches = []
    for index, token_record in enumerate(token_data.get("tokens", [])):
        if token_record_id(token_record) != remote_id:
            continue
        if endpoint and token_server_endpoint(token_record) != endpoint:
            continue
        matches.append((index, token_record))

    if not matches:
        return {
            "token_path": str(config.token_path),
            "remote": remote_id,
            "server": endpoint or "",
            "removed": "0",
            "status": "missing",
        }
    if len(matches) > 1 and not endpoint:
        available = ", ".join(sorted(token_server_endpoint(token_record) for _, token_record in matches))
        raise SystemExit(f"Multiple tokens with id '{remote_id}'. Pass --server. Available: {available}")

    removed_indexes = {index for index, _ in matches}
    first_record = matches[0][1]
    if not dry_run:
        token_data["tokens"] = [
            token_record
            for index, token_record in enumerate(token_data.get("tokens", []))
            if index not in removed_indexes
        ]
        write_token_store(config, token_data)

    return {
        "token_path": str(config.token_path),
        "remote": remote_id,
        "server": token_server_endpoint(first_record),
        "user": str(first_record.get("user", "")),
        "org": str(first_record.get("org", "")),
        "repo": str(first_record.get("repo", "")),
        "removed": str(len(matches)),
        "status": "dry-run" if dry_run else "removed",
    }


def remove_git_remote(repo_path: Path, remote_id: str, dry_run: bool = False) -> dict[str, str]:
    repo_root = git_repo_root(repo_path)
    if repo_root is None:
        raise SystemExit(f"Missing git repo at path: {repo_path}")

    remote_url = git_capture(repo_root, ["remote", "get-url", remote_id])
    if not remote_url:
        return {
            "repo_root": str(repo_root),
            "remote": remote_id,
            "removed": "0",
            "status": "missing",
        }

    if not dry_run:
        subprocess.run(
            ["git", "-C", str(repo_root), "remote", "remove", remote_id],
            check=True,
        )

    return {
        "repo_root": str(repo_root),
        "remote": remote_id,
        "url": safe_remote_url_label(remote_url),
        "removed": "1",
        "status": "dry-run" if dry_run else "removed",
    }


def run_tokens_remove(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    if args.target == "store":
        result = remove_token_from_store(config, args.remote_id, args.server, args.dry_run)
        print_token_action_result(result)
        return

    if args.target == "remote":
        result = remove_git_remote(resolve_launcher_repo_argument(args.repo), args.remote_id, args.dry_run)
        print_token_action_result(result)
        return

    if args.target == "both":
        repo_path = resolve_launcher_repo_argument(args.repo)
        if not args.dry_run:
            remove_token_from_store(config, args.remote_id, args.server, dry_run=True)
        remote_result = remove_git_remote(repo_path, args.remote_id, args.dry_run)
        print_token_action_result(remote_result)
        print()
        store_result = remove_token_from_store(config, args.remote_id, args.server, args.dry_run)
        print_token_action_result(store_result)
        return

    raise SystemExit(f"Unsupported tokens remove target: {args.target}")


def run_tokens_sync(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    repo_path = resolve_launcher_repo_argument(args.repo)
    if args.direction == "remote":
        result = sync_token_from_remote(config, repo_path, args.remote_id, args.dry_run)
        print_token_action_result(result)
        return
    if args.direction == "store":
        args.repo = str(repo_path)
        args.remote = args.remote_id
        args.token_name = args.remote_id
        args.auto_rename_origin = True
        run_tokens_write(config, args)
        return
    raise SystemExit(f"Unsupported tokens sync direction: {args.direction}")


def run_tokens_add(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    server_url = (args.server or config.git.base_url).rstrip("/")
    server_info = server_info_from_url(config, server_url)
    if server_info is None:
        raise SystemExit(f"Only http/https server URLs are supported: {server_url}")

    token_data = load_token_store(config)
    remote_id = args.token_id
    if args.remote and args.remote != remote_id:
        raise SystemExit("Token id must match --remote because tokens.json is keyed 1:1 by git remote name.")

    existing_token = find_token_record(token_data, server_info["endpoint"], remote_id)
    if existing_token is not None:
        raise SystemExit(f"Token already exists in tokens.json: {server_info['endpoint']}:{remote_id}")

    token_record = {
        "id": remote_id,
        "value": args.value or "",
        "server": server_record_from_info(server_info),
        "org": args.org or "",
        "repo": args.repo_name or "",
    }
    if args.user:
        token_record["user"] = args.user

    print(f"token_path\t{config.token_path}")
    print(f"server\t{server_info['endpoint']}")
    print(f"id\t{remote_id}")
    print(f"value\t{'set' if token_record['value'] else 'empty'}")
    print(f"remote\t{remote_id}")
    print(f"status\t{'dry-run' if args.dry_run else 'added'}")

    if args.dry_run:
        return

    token_data.setdefault("tokens", []).append(token_record)
    write_token_store(config, token_data)


def run_tokens_write(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    repo_path = resolve_repo_argument(args.repo)
    repo_root = git_repo_root(repo_path)
    if repo_root is None:
        raise SystemExit(f"Missing git repo at path: {repo_path}")

    if not args.remote:
        raise SystemExit("Pass --remote for tokens write.")

    token_data = load_token_store(config)
    existing_remote_url = git_capture(repo_root, ["remote", "get-url", args.remote])
    target_url = args.url or existing_remote_url
    target_endpoint = None
    if target_url:
        target_server_info = server_info_from_url(config, target_url)
        if target_server_info is None:
            raise SystemExit(f"Only http/https remote URLs are supported: {target_url}")
        target_endpoint = target_server_info["endpoint"]

    token_record = resolve_token_from_store(
        token_data,
        args.server or target_endpoint,
        args.token_name or args.remote,
        args.user,
    )
    server_endpoint = token_server_endpoint(token_record)
    remote_id = token_record_id(token_record)
    user_name = str(token_record.get("user", ""))
    token_value = str(token_record.get("value", ""))
    if not token_value:
        raise SystemExit(f"Token value is empty in tokens.json: {server_endpoint}:{remote_id}")
    if not user_name:
        raise SystemExit(f"Token user is empty in tokens.json: {server_endpoint}:{remote_id}")
    if not target_url:
        target_url = remote_url_from_token_record(token_record)
        if not target_url:
            raise SystemExit(
                f"Missing target remote URL and tokens.json has no org/repo for: {server_endpoint}:{remote_id}"
            )

    rename_candidate = None
    if getattr(args, "auto_rename_origin", False) and existing_remote_url is None:
        rename_candidate = find_origin_rename_candidate(config, repo_root, args.remote, target_url)
        if rename_candidate:
            _, existing_remote_url = rename_candidate

    final_url = credentialed_remote_url(target_url, user_name, token_value)
    status = "added" if existing_remote_url is None else "updated"
    existing_credentials = remote_credentials(existing_remote_url) if existing_remote_url else None
    if existing_credentials == (user_name, token_value):
        status = "renamed" if rename_candidate else "unchanged"
    elif existing_credentials is not None and not args.replace:
        raise SystemExit(
            f"Remote '{args.remote}' already has different credentials. "
            "Pass --replace to overwrite them."
        )

    print(f"repo_root\t{repo_root}")
    print(f"remote\t{args.remote}")
    if rename_candidate:
        print(f"renamed_from\t{rename_candidate[0]}")
    print(f"server\t{server_endpoint}")
    print(f"user\t{user_name}")
    print(f"id\t{remote_id}")
    print(f"status\t{status}")
    print(f"url\t{sanitize_remote_url(final_url)[0]}")

    if args.dry_run or status == "unchanged":
        return

    if rename_candidate:
        subprocess.run(
            ["git", "-C", str(repo_root), "remote", "rename", rename_candidate[0], args.remote],
            check=True,
        )
    set_git_remote(repo_root, args.remote, final_url)


def run_tokens_update(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    if args.remote_id:
        result = hydrate_token_from_api(config, args.remote_id, args.server, args.dry_run)
        print_token_action_result(result)
        return

    if args.from_source == "remotes":
        if args.dry_run:
            run_tokens_scan(config, args)
            return
        repo_path = resolve_repo_argument(args.repo)
        report = sync_tokens_from_repo(config, repo_path)
        repo_root, repo_servers = collect_repo_server_entries(config, repo_path)
        token_data = load_token_store(config, write_normalized=False)
        store_servers = store_servers_from_tokens(token_data)
        endpoint_names = sorted(set(repo_servers) | set(store_servers))
        print_token_item_rows(endpoint_names, repo_servers, store_servers, report["remote_rows"])
        return
    if args.from_source == "store":
        run_tokens_write(config, args)
        return
    raise SystemExit("Pass REMOTE_ID for API metadata refresh or use legacy --from remotes|store.")


def git_current_branch(config: WorkspaceConfig, repo_path: Path) -> str:
    branch_name = git_capture(repo_path, ["branch", "--show-current"])
    if branch_name:
        return branch_name

    head_ref = git_capture(repo_path, ["symbolic-ref", "--short", f"refs/remotes/{config.git.origin_remote}/HEAD"])
    if head_ref and "/" in head_ref:
        return head_ref.split("/", 1)[1]

    return config.git.fallback_branch


def fallback_source_repo_name(series_name: str, card_name: str) -> str:
    return "-".join([slug_value(series_name, "series"), slug_value(card_name, "card")])


def fallback_source_url(config: WorkspaceConfig, series_name: str, card_name: str) -> str:
    repo_name = fallback_source_repo_name(series_name, card_name)
    return f"{config.git.base_url}/{config.git.source_org}/{repo_name}.git"


def discover_source_url(config: WorkspaceConfig, repo_path: Path, series_name: str, card_name: str) -> str:
    source_url = git_capture(repo_path, ["remote", "get-url", config.git.origin_remote])
    if source_url:
        return source_url

    source_url = git_capture(repo_path, ["remote", "get-url", config.git.source_remote])
    if source_url:
        return source_url

    return fallback_source_url(config, series_name, card_name)


def repo_name_from_url(remote_url: str) -> str:
    repo_name = remote_url.rstrip("/").rsplit("/", 1)[-1]
    if repo_name.endswith(".git"):
        repo_name = repo_name[:-4]
    return slug_value(repo_name, "source repo")


def build_answer_repo_name(source_repo_name: str, class_name: str, date_value: str) -> str:
    return source_repo_name


def build_answer_url(config: WorkspaceConfig, answer_repo_name: str) -> str:
    return f"{config.git.base_url}/{config.git.answer_org}/{answer_repo_name}.git"


def answer_url_for_card(config: WorkspaceConfig, repo_name: str) -> str:
    return f"{config.git.base_url}/{config.git.answer_org}/{repo_name}.git"


def set_git_remote(repo_path: Path, remote_name: str, remote_url: str) -> str:
    existing_url = git_capture(repo_path, ["remote", "get-url", remote_name])
    if existing_url == remote_url:
        return "unchanged"
    if existing_url:
        subprocess.run(
            ["git", "-C", str(repo_path), "remote", "set-url", remote_name, remote_url],
            check=True,
        )
        return "updated"

    subprocess.run(
        ["git", "-C", str(repo_path), "remote", "add", remote_name, remote_url],
        check=True,
    )
    return "added"


def source_url_for_card(config: WorkspaceConfig, series_name: str, source_card_path: Path) -> str:
    source_url = git_capture(source_card_path, ["remote", "get-url", config.git.source_remote])
    if source_url:
        return source_url
    source_url = git_capture(source_card_path, ["remote", "get-url", config.git.origin_remote])
    if source_url:
        return source_url
    return f"{config.git.base_url}/{config.git.source_org}/{source_card_path.name}.git"


def source_url_for_card_info(config: WorkspaceConfig, card_info: CardInfo) -> str:
    card_config = config_for_card(config, card_info)
    if card_info.source_path:
        return source_url_for_card(card_config, card_info.series_id, card_info.source_path)
    return f"{card_config.git.base_url}/{card_info.source_org}/{card_info.repo}.git"


def ensure_answer_token_from_source(
    config: WorkspaceConfig,
    source_remote: str,
    answer_remote: str,
    answer_url: str,
    dry_run: bool = False,
) -> tuple[str, dict]:
    token_data = load_token_store(config, write_normalized=not dry_run)
    answer_server_info = server_info_from_url(config, answer_url)
    if answer_server_info is None:
        raise SystemExit(f"Only http/https remote URLs are supported: {answer_url}")
    source_token = resolve_token_from_store(token_data, answer_server_info["endpoint"], source_remote, None)
    if dry_run:
        return "dry-run", source_token
    status = register_token(
        token_data,
        answer_server_info,
        answer_remote,
        str(source_token.get("user", "")),
        str(source_token.get("value", "")),
        hydrate=False,
        reset_api_metadata=True,
    )
    write_token_store(config, token_data)
    return status, source_token


def workspace_card_path(config: WorkspaceConfig, series_name: str, source_card_path: Path) -> Path:
    return workspace_series_path(config, series_name) / source_card_path.name


def workspace_card_path_for_info(config: WorkspaceConfig, card_info: CardInfo) -> Path:
    return workspace_series_path(config, card_info.series_id) / card_info.repo


def resolve_workspace_card_path(config: WorkspaceConfig, series_name: str, card_name: str) -> Path:
    card_info = resolve_card_info(config, series_name, card_name)
    card_path = workspace_card_path_for_info(config, card_info)
    if not card_path.is_dir():
        raise SystemExit(f"Missing workspace card: {card_path}. Run: ./rvctl series cards fetch {series_name} {card_name}")
    return card_path


def fetch_card_to_workspace(config: WorkspaceConfig, series_name: str, card_name: str, dry_run: bool = False) -> dict[str, str]:
    card_info = resolve_card_info(config, series_name, card_name)
    card_config = config_for_card(config, card_info)
    repo_name = card_info.repo
    target_path = workspace_card_path_for_info(config, card_info)
    source_url = source_url_for_card_info(config, card_info)
    answer_url = answer_url_for_card(card_config, repo_name)
    source_branch = card_info.branch

    if not dry_run:
        target_path.parent.mkdir(parents=True, exist_ok=True)
        if target_path.exists() and git_repo_root(target_path) is None:
            raise SystemExit(f"Workspace card path exists but is not a git repo: {target_path}")
        cloned = False
        if not target_path.exists():
            subprocess.run(["git", "clone", source_url, str(target_path)], check=True)
            cloned = True
        origin_url = git_capture(target_path, ["remote", "get-url", card_config.git.origin_remote])
        if origin_url and git_capture(target_path, ["remote", "get-url", card_config.git.source_remote]) is None:
            subprocess.run(
                ["git", "-C", str(target_path), "remote", "rename", card_config.git.origin_remote, card_config.git.source_remote],
                check=True,
            )
        set_git_remote(target_path, card_config.git.source_remote, source_url)
        subprocess.run(["git", "-C", str(target_path), "fetch", card_config.git.source_remote, source_branch], check=True)
        current_branch = git_current_branch(card_config, target_path)
        if cloned or current_branch == source_branch:
            if current_branch != source_branch:
                local_branch_exists = subprocess.run(
                    ["git", "-C", str(target_path), "rev-parse", "--verify", "--quiet", source_branch],
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                    check=False,
                ).returncode == 0
                if local_branch_exists:
                    subprocess.run(["git", "-C", str(target_path), "switch", source_branch], check=True)
                else:
                    subprocess.run(
                        ["git", "-C", str(target_path), "switch", "--track", "-c", source_branch, f"{card_config.git.source_remote}/{source_branch}"],
                        check=True,
                    )
            subprocess.run(["git", "-C", str(target_path), "pull", "--ff-only", card_config.git.source_remote, source_branch], check=True)

    token_status, source_token = ensure_answer_token_from_source(
        card_config,
        card_config.git.source_remote,
        card_config.git.answer_remote,
        answer_url,
        dry_run=dry_run,
    )
    answer_remote_status = "dry-run"
    if not dry_run:
        answer_remote_url = credentialed_remote_url(
            answer_url,
            str(source_token.get("user", "")),
            str(source_token.get("value", "")),
        )
        answer_remote_status = set_git_remote(target_path, card_config.git.answer_remote, answer_remote_url)

    return {
        "series": series_name,
        "card": card_info.card_id,
        "repo": repo_name,
        "source": source_url,
        "workspace": str(target_path),
        "source_remote": card_config.git.source_remote,
        "source_branch": source_branch,
        "answer_remote": card_config.git.answer_remote,
        "answer_org": card_config.git.answer_org,
        "answer_repo": repo_name,
        "answer_url": answer_url,
        "token": card_config.git.answer_remote,
        "token_status": token_status,
        "answer_remote_status": answer_remote_status,
        "status": "dry-run" if dry_run else "ready",
    }


def print_key_values(values: dict[str, str]) -> None:
    for key, value in values.items():
        print(f"{key}\t{value}")


def task_name_from_path(path: Path) -> str:
    if path.is_file():
        return path.stem
    return path.name


def task_entries(card_path: Path) -> list[TaskEntry]:
    entries: dict[str, TaskEntry] = {}
    task_root = card_path / "src" / "tasks"
    if task_root.is_dir():
        for path in sorted(task_root.iterdir()):
            name = task_name_from_path(path)
            if (path.is_file() or path.is_dir()) and name.startswith("task"):
                entries[name] = TaskEntry(name=name, path=path)

    for path in sorted(card_path.iterdir()):
        name = task_name_from_path(path)
        if (path.is_file() or path.is_dir()) and name.startswith("task"):
            entries.setdefault(name, TaskEntry(name=name, path=path))

    return [entries[name] for name in sorted(entries)]


def task_names(card_path: Path) -> list[str]:
    return [entry.name for entry in task_entries(card_path)]


def task_branch_part(task_name: str) -> str:
    match = re.search(r"(?i)task[-_]?(\d+)", task_name)
    if match is None:
        match = re.fullmatch(r"(?i)t(\d+)", task_name.strip())
    if match is None:
        match = re.search(r"(\d+)", task_name)
    if match is None:
        raise SystemExit(f"Task name does not contain a task number: {task_name}")
    return f"T{int(match.group(1))}"


def resolve_submission_task(config: WorkspaceConfig, card_path: Path, task_arg: str | None) -> str:
    task_selector = task_arg or config.defaults.get("task")
    if not task_selector:
        raise SystemExit("Missing task selector. Pass --task or set defaults.task in workspace.json.")

    names = task_names(card_path)
    if not names:
        return task_selector

    matches = [name for name in names if name == task_selector]
    if not matches:
        try:
            selector_part = task_branch_part(task_selector)
        except SystemExit:
            selector_part = ""
        number_matches = []
        for name in names:
            try:
                if selector_part and task_branch_part(name) == selector_part:
                    number_matches.append(name)
            except SystemExit:
                pass
        matches = number_matches
    if not matches:
        matches = [name for name in names if task_selector in name]

    if not matches:
        raise SystemExit(f"Missing task '{task_selector}' in card: {card_path}")
    if len(matches) > 1:
        raise SystemExit(f"Ambiguous task '{task_selector}'. Matches: {', '.join(matches)}")
    return matches[0]


def store_user_for_submission(
    config: WorkspaceConfig,
    source_url: str,
    answer_url: str,
    required: bool = True,
) -> str:
    token_data = load_token_store(config)
    endpoints = []
    for remote_url in [source_url, answer_url, config.git.base_url]:
        server_info = server_info_from_url(config, remote_url)
        if server_info is None:
            continue
        endpoint = server_info["endpoint"]
        if endpoint not in endpoints:
            endpoints.append(endpoint)

    for endpoint in endpoints:
        for remote_id in [config.git.source_remote, config.git.answer_remote]:
            token_record = find_token_record(token_data, endpoint, remote_id)
            if token_record and token_record.get("user"):
                return slug_value(str(token_record["user"]), "token user")

    users = {
        str(token_record.get("user", ""))
        for token_record in token_data.get("tokens", [])
        if token_server_endpoint(token_record) in endpoints and token_record.get("user")
    }
    if len(users) == 1:
        return slug_value(next(iter(users)), "token user")

    if required:
        raise SystemExit("Missing --nick and no unique user in tokens.json for the answer branch.")
    return ""


def submission_branch_name(user_name: str, task_name: str) -> str:
    return branch_value(f"{slug_value(user_name, 'student user')}{task_branch_part(task_name)}a")


def set_branch_upstream(repo_path: Path, branch_name: str, remote_name: str, dry_run: bool = False) -> str:
    expected_merge = f"refs/heads/{branch_name}"
    current_remote = git_capture(repo_path, ["config", "--get", f"branch.{branch_name}.remote"])
    current_merge = git_capture(repo_path, ["config", "--get", f"branch.{branch_name}.merge"])
    if current_remote == remote_name and current_merge == expected_merge:
        return "unchanged"
    if dry_run:
        return "dry-run"
    subprocess.run(["git", "-C", str(repo_path), "config", f"branch.{branch_name}.remote", remote_name], check=True)
    subprocess.run(["git", "-C", str(repo_path), "config", f"branch.{branch_name}.merge", expected_merge], check=True)
    return "updated" if current_remote or current_merge else "set"


def print_card_tasks(config: WorkspaceConfig, series_name: str, card_name: str) -> None:
    card_path = resolve_workspace_card_path(config, series_name, card_name)
    for entry in task_entries(card_path):
        print(f"{series_name}\t{card_label_from_repo_name(card_path.name)}\t{entry.name}\t{entry.path}")


def print_card_task(config: WorkspaceConfig, series_name: str, card_name: str, task_name: str) -> None:
    card_path = resolve_workspace_card_path(config, series_name, card_name)
    matches = [entry for entry in task_entries(card_path) if entry.name == task_name or task_name in entry.name]
    if not matches:
        raise SystemExit(f"Missing task '{task_name}' in card: {card_path}")
    if len(matches) > 1:
        raise SystemExit(f"Ambiguous task '{task_name}'. Matches: {', '.join(entry.name for entry in matches)}")
    task_entry = matches[0]
    print(f"series\t{series_name}")
    print(f"card\t{card_label_from_repo_name(card_path.name)}")
    print(f"task\t{task_entry.name}")
    print(f"card_path\t{card_path}")
    print(f"task_path\t{task_entry.path}")


def switch_card_task(
    config: WorkspaceConfig,
    series_name: str,
    card_name: str,
    task_selector: str,
    branch_override: str | None = None,
    dry_run: bool = False,
) -> dict[str, str]:
    card_info = resolve_card_info(config, series_name, card_name)
    card_config = config_for_card(config, card_info)
    card_path = resolve_workspace_card_path(config, series_name, card_name)
    task_name = resolve_submission_task(config, card_path, task_selector)
    source_url = discover_source_url(card_config, card_path, series_name, card_name)
    source_repo_name = repo_name_from_url(source_url)
    answer_url = build_answer_url(card_config, source_repo_name)
    student_user = store_user_for_submission(card_config, source_url, answer_url, required=branch_override is None)
    branch_name = branch_value(branch_override) if branch_override else submission_branch_name(student_user, task_name)
    current_branch = git_current_branch(card_config, card_path)
    branch_exists = subprocess.run(
        ["git", "-C", str(card_path), "rev-parse", "--verify", "--quiet", branch_name],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    ).returncode == 0

    token_status = "dry-run"
    answer_remote_status = "dry-run"
    if not dry_run:
        token_status, source_token = ensure_answer_token_from_source(
            card_config,
            card_config.git.source_remote,
            card_config.git.answer_remote,
            answer_url,
        )
        answer_remote_status = set_git_remote(
            card_path,
            card_config.git.answer_remote,
            credentialed_remote_url(
                answer_url,
                str(source_token.get("user", "")),
                str(source_token.get("value", "")),
            ),
        )

    if dry_run:
        status = "dry-run"
    elif current_branch == branch_name:
        status = "unchanged"
    elif branch_exists:
        subprocess.run(["git", "-C", str(card_path), "switch", branch_name], check=True)
        status = "switched"
    else:
        subprocess.run(["git", "-C", str(card_path), "switch", "-c", branch_name], check=True)
        status = "created"
    upstream_status = set_branch_upstream(card_path, branch_name, card_config.git.answer_remote, dry_run=dry_run)

    return {
        "series": series_name,
        "card": card_info.card_id,
        "task": task_name,
        "student_user": student_user,
        "branch": branch_name,
        "answer_remote": card_config.git.answer_remote,
        "answer_url": answer_url,
        "token_status": token_status,
        "answer_remote_status": answer_remote_status,
        "upstream": f"{card_config.git.answer_remote}/{branch_name}",
        "upstream_status": upstream_status,
        "workspace": str(card_path),
        "status": status,
    }


def print_series_details(config: WorkspaceConfig, series_name: str) -> None:
    config.series_root.mkdir(parents=True, exist_ok=True)
    source_path = source_series_path(config, series_name)
    workspace_path = workspace_series_path(config, series_name)
    cards = source_card_infos(config, series_name)
    if not cards and not workspace_path.is_dir():
        raise SystemExit(f"Missing series: {series_name}")
    print(f"series\t{series_name}")
    print(f"source_path\t{source_path if source_path.is_dir() else ''}")
    print(f"manifest\t{config.workspace_info_path if workspace_series_manifest(config, series_name) else ''}")
    print(f"workspace_path\t{workspace_path}")
    print(f"cards_count\t{len(cards)}")
    print(f"cards\t{' '.join(card.card_id for card in cards)}")


def print_card_details(config: WorkspaceConfig, series_name: str, card_name: str) -> None:
    card_info = resolve_card_info(config, series_name, card_name)
    card_config = config_for_card(config, card_info)
    target_path = workspace_card_path_for_info(config, card_info)
    repo_name = card_info.repo
    print(f"series\t{series_name}")
    print(f"card\t{card_info.card_id}")
    print(f"repo\t{repo_name}")
    print(f"source_path\t{card_info.source_path or ''}")
    print(f"workspace_path\t{target_path}")
    print(f"source_remote\t{card_config.git.source_remote}")
    print(f"source_url\t{source_url_for_card_info(config, card_info)}")
    print(f"source_branch\t{card_info.branch}")
    print(f"answer_remote\t{card_config.git.answer_remote}")
    print(f"answer_url\t{answer_url_for_card(card_config, repo_name)}")
    if target_path.is_dir():
        print(f"tasks\t{' '.join(task_names(target_path))}")


def available_series_hint(config: WorkspaceConfig) -> str:
    manifest_names = list(workspace_series_manifests(config))
    if manifest_names:
        names = sorted(manifest_names)
    else:
        names = sorted({path.name for path in source_series_directories(config)} | {path.name for path in series_directories(config)})
    return ", ".join(names)


def use_series_default(config: WorkspaceConfig, series_name: str) -> dict[str, str]:
    series_id = slug_value(series_name, "series")
    if not series_exists(config, series_id):
        hint = available_series_hint(config)
        suffix = f" Available: {hint}" if hint else ""
        raise SystemExit(f"Missing series '{series_name}'.{suffix}")
    return update_config_defaults(config, {"series": series_id})


def use_card_default(config: WorkspaceConfig, selector: list[str]) -> dict[str, str]:
    if len(selector) == 1:
        series_name = config.defaults.get("series")
        if not series_name:
            raise SystemExit("Missing default series. Use: ./rvctl series use inf")
        card_name = selector[0]
        updates_series = False
    elif len(selector) == 2:
        series_name = slug_value(selector[0], "series")
        card_name = selector[1]
        updates_series = True
    else:
        raise SystemExit("Use: ./rvctl card use <card> or ./rvctl card use <series> <card>")

    if not series_exists(config, series_name):
        hint = available_series_hint(config)
        suffix = f" Available: {hint}" if hint else ""
        raise SystemExit(f"Missing series '{series_name}'.{suffix}")

    card_info = resolve_card_info(config, series_name, card_name)
    updates = {"card": card_info.card_id}
    if updates_series:
        updates = {"series": series_name, "card": card_info.card_id}
    return update_config_defaults(config, updates)


def run_series(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    if args.series_command == "list":
        print_series(config)
        return
    if args.series_command == "use":
        print_key_values(use_series_default(config, args.series))
        return
    if args.series_command == "show":
        print_series_details(config, args.series)
        return
    if args.series_command == "fetch":
        for card_info in source_card_infos(config, args.series):
            result = fetch_card_to_workspace(config, args.series, card_info.card_id, dry_run=args.dry_run)
            print_key_values(result)
            print()
        return
    if args.series_command != "cards":
        raise SystemExit(f"Unsupported series command: {args.series_command}")

    if args.cards_command == "list":
        print_cards(config, args.series)
        return
    if args.cards_command == "show":
        print_card_details(config, args.series, args.card)
        return
    if args.cards_command == "fetch":
        print_key_values(fetch_card_to_workspace(config, args.series, args.card, dry_run=args.dry_run))
        return
    if args.cards_command == "submission":
        print_submission_plan(config, args)
        return
    if args.cards_command != "tasks":
        raise SystemExit(f"Unsupported series cards command: {args.cards_command}")

    if args.tasks_command == "list":
        print_card_tasks(config, args.series, args.card)
        return
    if args.tasks_command == "show":
        print_card_task(config, args.series, args.card, args.task)
        return
    if args.tasks_command == "switch":
        print_key_values(switch_card_task(config, args.series, args.card, args.task, args.branch, args.dry_run))
        return
    raise SystemExit(f"Unsupported series cards tasks command: {args.tasks_command}")


def run_card(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    if args.card_command == "use":
        print_key_values(use_card_default(config, args.selector))
        return
    raise SystemExit(f"Unsupported card command: {args.card_command}")


def resolve_task_command_selector(
    config: WorkspaceConfig,
    values: list[str],
    needs_task: bool = False,
) -> tuple[str, str, str | None]:
    if not needs_task:
        if len(values) > 2:
            raise SystemExit("Pass at most '<series> <card>' for tasks list.")
        series_arg = values[0] if values else None
        card_arg = values[1] if len(values) == 2 else None
        series_name, card_name = resolve_selector(config, series_arg, card_arg)
        return series_name, card_name, None

    if not values:
        raise SystemExit("Missing task. Use: ./rvctl tasks switch 4")
    if len(values) > 3:
        raise SystemExit("Pass '<task>', '<card> <task>' or '<series> <card> <task>'.")
    if len(values) == 1:
        series_name, card_name = resolve_selector(config, None, None)
        return series_name, card_name, values[0]
    if len(values) == 2:
        series_name, card_name = resolve_selector(config, values[0], None)
        return series_name, card_name, values[1]
    series_name, card_name = resolve_selector(config, values[0], values[1])
    return series_name, card_name, values[2]


def run_tasks(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    if args.tasks_command == "list":
        series_name, card_name, _ = resolve_task_command_selector(config, args.selector)
        print_card_tasks(config, series_name, card_name)
        return
    if args.tasks_command == "show":
        series_name, card_name, task_name = resolve_task_command_selector(config, args.selector, needs_task=True)
        print_card_task(config, series_name, card_name, task_name or "")
        return
    if args.tasks_command == "switch":
        series_name, card_name, task_name = resolve_task_command_selector(config, args.selector, needs_task=True)
        print_key_values(switch_card_task(config, series_name, card_name, task_name or "", args.branch, args.dry_run))
        return
    raise SystemExit(f"Unsupported tasks command: {args.tasks_command}")


def print_submission_plan(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    series_name, card_name = resolve_selector(config, args.series, args.card)
    card_info = resolve_card_info(config, series_name, card_name)
    card_config = config_for_card(config, card_info)
    selector = f"{series_name}/{card_name}"
    card_path = resolve_card_path(config, series_name, card_name)
    sync_tokens_from_repo(card_config, card_path)
    class_name = slug_value(args.class_name, "class")
    source_url = args.source_url or discover_source_url(card_config, card_path, series_name, card_name)
    source_repo_name = repo_name_from_url(source_url)
    answer_repo_name = build_answer_repo_name(source_repo_name, class_name, args.date)
    answer_url = build_answer_url(card_config, answer_repo_name)
    source_branch = git_current_branch(card_config, card_path)
    task_name = resolve_submission_task(config, card_path, args.task)
    student_user = (
        slug_value(args.nick, "student user")
        if args.nick
        else store_user_for_submission(card_config, source_url, answer_url, required=args.branch is None)
    )
    branch_name = branch_value(args.branch) if args.branch else submission_branch_name(student_user, task_name)

    print(f"selector\t{selector}")
    print(f"card_path\t{card_path}")
    print(f"task\t{task_name}")
    print(f"source_repo\t{card_config.git.source_org}/{source_repo_name}")
    print(f"source_remote\t{card_config.git.source_remote}")
    print(f"source_url\t{source_url}")
    print(f"source_branch\t{source_branch}")
    print(f"answer_remote\t{card_config.git.answer_remote}")
    print(f"answer_repo\t{card_config.git.answer_org}/{answer_repo_name}")
    print(f"answer_url\t{answer_url}")
    print(f"student_user\t{student_user}")
    print(f"student_branch\t{branch_name}")

    if args.apply:
        source_status = set_git_remote(card_path, card_config.git.source_remote, source_url)
        token_status, source_token = ensure_answer_token_from_source(
            card_config,
            card_config.git.source_remote,
            card_config.git.answer_remote,
            answer_url,
        )
        answer_status = set_git_remote(
            card_path,
            card_config.git.answer_remote,
            credentialed_remote_url(
                answer_url,
                str(source_token.get("user", "")),
                str(source_token.get("value", "")),
            ),
        )
        print(f"apply_{card_config.git.source_remote}\t{source_status}")
        print(f"apply_token_{card_config.git.answer_remote}\t{token_status}")
        print(f"apply_{card_config.git.answer_remote}\t{answer_status}")

    print("commands")
    print(f"  git -C {shlex.quote(str(card_path))} remote add {card_config.git.source_remote} {shlex.quote(source_url)}")
    print(f"  git -C {shlex.quote(str(card_path))} remote add {card_config.git.answer_remote} {shlex.quote(answer_url)}")
    print(f"  git -C {shlex.quote(str(card_path))} fetch {card_config.git.source_remote} {shlex.quote(source_branch)}")
    print(f"  git -C {shlex.quote(str(card_path))} switch -c {shlex.quote(branch_name)}")
    print(f"  git -C {shlex.quote(str(card_path))} push -u {card_config.git.answer_remote} {shlex.quote(branch_name)}")


def tmux_target_exists(session_name: str) -> bool:
    result = subprocess.run(
        ["tmux", "has-session", "-t", session_name],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    )
    return result.returncode == 0


def run_tmux_container(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    series_name, card_name = resolve_selector(config, args.series, args.card)
    selector = f"{series_name}/{card_name}"
    card_path = resolve_card_path(config, series_name, card_name)
    profile = normalize_env_profile(args.profile)
    task_name = resolve_submission_task(config, card_path, None)
    instance = args.instance or default_env_instance(profile, series_name, card_name, task_name, "shell")
    window_name = args.window or config.defaults.get("tmux_window", "rv")
    session_name = args.session or slug_value(f"stem-{instance}", "tmux session")

    if tmux_target_exists(session_name):
        raise SystemExit(
            f"tmux session already exists: {session_name}\n"
            "Use another --session name or remove the existing session first."
        )

    # Compatibility command: keep the outer host tmux session, but delegate
    # container creation/mounts/labels entirely to the canonical rootless
    # ./stem runtime. Do not maintain a second Docker Compose lifecycle here.
    shell_command, details = env_action_command(
        config,
        profile,
        "shell",
        series_name=series_name,
        card_name=card_name,
        task_name=task_name,
        instance=instance,
    )
    tmux_command = [
        "tmux",
        "new-session",
        "-d",
        "-s",
        session_name,
        "-n",
        window_name,
        f"bash -lc {shlex.quote(shell_command)}",
    ]

    if args.dry_run:
        print("Selector:", selector)
        print("Card path:", card_path)
        print("Tools root:", config.tools_root)
        print("Runtime: rootless stem/Podman")
        print("Instance:", instance)
        print("Pane 0 command:")
        print(shell_command)
        print("tmux command:")
        print(shlex.join(tmux_command))
        print_env_plan(details)
        return

    subprocess.run(tmux_command, check=True)

    if args.attach:
        subprocess.run(["tmux", "attach", "-t", session_name], check=True)
        return

    print(f"Created tmux session: {session_name}")
    print(f"Window: {window_name}")
    print("Pane 0: container shell")
    print(f"Attach with: tmux attach -t {session_name}")


def normalize_env_profile(profile: str) -> str:
    normalized = ENV_PROFILE_ALIASES.get(profile, profile)
    if normalized not in ENV_PROFILES:
        available = ", ".join(sorted(ENV_PROFILES))
        raise SystemExit(f"Unsupported environment profile '{profile}'. Available: {available}")
    return normalized


def env_tool_command_path(config: WorkspaceConfig) -> Path:
    stem_script = config.tools_root / "stem"
    return stem_script if stem_script.is_file() else config.tools_root / "rv"


def ensure_env_tool(config: WorkspaceConfig) -> Path:
    env_script = env_tool_command_path(config)
    if env_script.is_file():
        return env_script
    candidates = "\n".join(f"  - {path}" for path in config.tools_root_candidates)
    raise SystemExit(
        "Missing environment launcher script 'stem' (or compatible 'rv').\n"
        f"Checked:\n{candidates}\n"
        "Run: ./stemctl env sync"
    )


def env_sync_target(config: WorkspaceConfig) -> Path:
    return config.workspace_root / "tools" / "rv32i-hazard3-student-env"


def sync_env_tool(config: WorkspaceConfig, dry_run: bool = False) -> dict[str, str]:
    target = env_sync_target(config)
    if dry_run:
        command = "pull" if (target / ".git").is_dir() else "clone"
        return {
            "env_tool_path": str(target),
            "env_tool_url": config.env_tool_url,
            "command": command,
            "status": "dry-run",
        }
    target.parent.mkdir(parents=True, exist_ok=True)
    if (target / ".git").is_dir():
        subprocess.run(["git", "-C", str(target), "pull", "--ff-only"], check=True)
        config.tools_root = target
        return {
            "env_tool_path": str(target),
            "env_tool_url": config.env_tool_url,
            "command": "pull",
            "status": "ready",
        }
    if target.exists():
        raise SystemExit(f"Environment tool path exists but is not a git repo: {target}")
    subprocess.run(["git", "clone", config.env_tool_url, str(target)], check=True)
    config.tools_root = target
    return {
        "env_tool_path": str(target),
        "env_tool_url": config.env_tool_url,
        "command": "clone",
        "status": "ready",
    }


def task_selector_like(value: str) -> bool:
    return bool(re.fullmatch(r"(?i)(task[-_]?)?\d+.*|t\d+", value.strip()))


def resolve_env_target(
    config: WorkspaceConfig,
    values: list[str],
    task_override: str | None = None,
) -> tuple[str, str, str]:
    if len(values) > 3:
        raise SystemExit("Pass '<task>', '<card> [task]' or '<series> <card> [task]'.")

    if not values:
        series_name, card_name = resolve_selector(config, None, None)
        card_path = resolve_workspace_card_path(config, series_name, card_name)
        task_name = resolve_submission_task(config, card_path, task_override)
        return series_name, card_name, task_name

    if len(values) == 1:
        if task_selector_like(values[0]):
            series_name, card_name = resolve_selector(config, None, None)
            task_selector = task_override or values[0]
        else:
            series_name, card_name = resolve_selector(config, values[0], None)
            task_selector = task_override
        card_path = resolve_workspace_card_path(config, series_name, card_name)
        task_name = resolve_submission_task(config, card_path, task_selector)
        return series_name, card_name, task_name

    if len(values) == 2:
        if task_selector_like(values[1]):
            series_name, card_name = resolve_selector(config, values[0], None)
            task_selector = task_override or values[1]
        else:
            series_name, card_name = resolve_selector(config, values[0], values[1])
            task_selector = task_override
        card_path = resolve_workspace_card_path(config, series_name, card_name)
        task_name = resolve_submission_task(config, card_path, task_selector)
        return series_name, card_name, task_name

    series_name, card_name = resolve_selector(config, values[0], values[1])
    card_path = resolve_workspace_card_path(config, series_name, card_name)
    task_name = resolve_submission_task(config, card_path, task_override or values[2])
    return series_name, card_name, task_name


def split_pane_selector(values: list[str], explicit_pane: str | None = None) -> tuple[list[str], str | None]:
    if "pane" not in values:
        return values, explicit_pane
    if explicit_pane:
        raise SystemExit("Pass pane target either as '--pane TARGET' or as 'pane TARGET', not both.")

    index = values.index("pane")
    pane_values = values[index + 1 :]
    if len(pane_values) != 1:
        raise SystemExit("Use: pane TARGET, for example 'pane 0' or 'pane node:0'.")
    return values[:index], pane_values[0]


def tmux_pane_target(raw_target: str) -> str:
    target = raw_target.strip()
    if not target:
        raise SystemExit("Empty tmux pane target.")
    if re.fullmatch(r"\d+", target):
        return f":.{target}"
    if re.fullmatch(r"%\d+", target):
        return target
    window_pane_match = re.fullmatch(r"([^:.]+):(\d+)", target)
    if window_pane_match:
        window_name, pane_index = window_pane_match.groups()
        return f":{window_name}.{pane_index}"
    return target


def send_command_to_tmux_pane(shell_command: str, raw_target: str, dry_run: bool = False) -> dict[str, str]:
    target = tmux_pane_target(raw_target)
    details = {
        "pane": raw_target,
        "tmux_target": target,
        "tmux_send": f"tmux send-keys -t {shlex.quote(target)} -l {shlex.quote(shell_command)}; tmux send-keys -t {shlex.quote(target)} C-m",
    }
    if dry_run:
        return details

    resolved = subprocess.run(
        ["tmux", "display-message", "-p", "-t", target, "#{session_name}:#{window_name}.#{pane_index}"],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        check=False,
    )
    if resolved.returncode != 0:
        raise SystemExit(f"Missing tmux pane target '{raw_target}' ({target}): {resolved.stderr.strip()}")

    subprocess.run(["tmux", "send-keys", "-t", target, "-l", shell_command], check=True)
    subprocess.run(["tmux", "send-keys", "-t", target, "C-m"], check=True)
    details["tmux_resolved"] = resolved.stdout.strip()
    return details


def default_env_instance(profile: str, series_name: str, card_name: str, task_name: str, action: str) -> str:
    task_part = task_branch_part(task_name).lower()
    del action
    return slug_value(f"{profile}-{series_name}-{card_name}-{task_part}", "environment instance")


def env_action_command(
    config: WorkspaceConfig,
    profile: str,
    action: str,
    series_name: str | None = None,
    card_name: str | None = None,
    task_name: str | None = None,
    instance: str | None = None,
    editor: str | None = None,
    target: str | None = None,
    device: str | None = None,
    deploy_backend: str | None = None,
    allow_existing_firmware: bool = False,
) -> tuple[str, dict[str, str]]:
    profile = normalize_env_profile(profile)
    env_script = ensure_env_tool(config)
    selected_target = target or PROFILE_DEFAULT_TARGETS[profile]
    env_vars = {
        "STEM_WORKSPACE_ROOT": str(config.workspace_root),
        "STEM_CARDS_ROOT": str(config.series_root),
        "STEM_SOCKET_ROOT": str(config.socket_root),
        "STEM_PROFILE": profile,
        "STEM_TARGET": selected_target,
        "WORKSPACE_ROOT": str(config.workspace_root),
        "RV_CARDS_ROOT": str(config.series_root),
        "RV_PROFILE": profile,
    }
    command_args = ["bash", f"./{env_script.name}"]

    card_path: Path | None = None
    selector = ""
    if series_name and card_name:
        card_path = resolve_workspace_card_path(config, series_name, card_name)
        selector = f"{series_name}/{card_name}"
        env_vars.update(
            {
                "STEM_REPO_PATH": str(card_path),
                "STEM_REPO_HOST": str(card_path),
                "STEM_CARD": card_path.name,
                "RV_REPO_PATH": str(card_path),
                "RV_REPO_HOST": str(card_path),
                "RV_CARD": card_path.name,
            }
        )

    if instance:
        env_vars["STEM_INSTANCE"] = instance
        env_vars["RV_INSTANCE"] = instance
    elif series_name and card_name and task_name:
        generated_instance = default_env_instance(profile, series_name, card_name, task_name, action)
        env_vars["STEM_INSTANCE"] = generated_instance
        env_vars["RV_INSTANCE"] = generated_instance

    if editor:
        env_vars["STEM_EDITOR"] = editor
        env_vars["RV_EDITOR"] = editor

    if device:
        env_vars["STEM_PROBE_DEVICE"] = device
    if deploy_backend:
        env_vars["STEM_DEPLOY_BACKEND"] = deploy_backend

    if action in {"build", "test", "run", "debug", "deploy"}:
        if not task_name:
            raise SystemExit(f"{action} requires a task selector.")
        command_args.extend([action, "--target", selected_target, task_name])
        if device:
            command_args.extend(["--device", device])
        if action == "deploy" and deploy_backend:
            command_args.extend(["--backend", deploy_backend])
        if allow_existing_firmware:
            command_args.append("--allow-existing-firmware")
    elif action in {"shell", "start", "status", "attach", "stop", "rm"}:
        command_args.extend([action, "--target", selected_target])
    else:
        raise SystemExit(f"Unsupported environment action: {action}")

    env_prefix = " ".join(f"{key}={shlex.quote(value)}" for key, value in env_vars.items())
    shell_command = " && ".join(
        [
            f"cd {shlex.quote(str(env_script.parent))}",
            " ".join([env_prefix, *[shlex.quote(part) for part in command_args]]),
        ]
    )
    details = {
        "profile": profile,
        "service": ENV_PROFILES[profile]["service"],
        "target": selected_target,
        "selector": selector,
        "task": task_name or "",
        "card_path": str(card_path or ""),
        "tools_root": str(config.tools_root),
        "instance": env_vars.get("RV_INSTANCE", ""),
        "device": device or "",
        "backend": deploy_backend or "",
        "command": shell_command,
    }
    return shell_command, details


def print_env_plan(details: dict[str, str]) -> None:
    keys = [
        "profile",
        "service",
        "target",
        "selector",
        "task",
        "card_path",
        "tools_root",
        "instance",
        "device",
        "backend",
        "pane",
        "tmux_target",
        "tmux_resolved",
        "tmux_send",
        "command",
    ]
    for key in keys:
        if key == "command" or details.get(key):
            print(f"{key}\t{details.get(key, '')}")


def run_env(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    if args.env_command == "list":
        print("profile\tservice\timage\tstatus\tpurpose")
        for profile, info in ENV_PROFILES.items():
            print(
                f"{profile}\t{info['service']}\t{info['image']}\t"
                f"{'ready' if env_tool_command_path(config).is_file() else 'missing'}\t{info['purpose']}"
            )
        print(f"tools_root\t{config.tools_root}")
        print(f"env_tool_url\t{config.env_tool_url}")
        return
    if args.env_command == "sync":
        print_key_values(sync_env_tool(config, dry_run=args.dry_run))
        return
    if args.env_command == "sources":
        series_name, card_name = resolve_selector(config, args.series, args.card)
        card_path = resolve_workspace_card_path(config, series_name, card_name)
        env_script = ensure_env_tool(config)
        preparer = env_script.parent / "scripts" / "prepare-workspace-sources.py"
        if not preparer.is_file():
            raise SystemExit(
                f"Environment tool does not provide pinned workspace sources: {preparer}. "
                "Run: ./stemctl env sync"
            )
        mode = "check" if args.check else "prepare"
        command = [str(preparer), mode, str(card_path)]
        if args.dry_run:
            print(f"series\t{series_name}")
            print(f"card\t{card_name}")
            print(f"workspace\t{card_path}")
            print(f"command\t{shlex.join(command)}")
            return
        subprocess.run(command, check=True)
        return
    if args.env_command == "build":
        profile = normalize_env_profile(args.profile)
        env_script = ensure_env_tool(config)
        shell_command = " && ".join(
            [
                f"cd {shlex.quote(str(env_script.parent))}",
                " ".join(
                    [
                        f"STEM_PROFILE={shlex.quote(profile)}",
                        f"STEM_WORKSPACE_ROOT={shlex.quote(str(config.workspace_root))}",
                        f"STEM_SOCKET_ROOT={shlex.quote(str(config.socket_root))}",
                        "bash",
                        shlex.quote(f"./{env_script.name}"),
                        "profile-build",
                        shlex.quote(profile),
                    ]
                ),
            ]
        )
        if args.dry_run:
            print(f"profile\t{profile}")
            print(f"command\t{shell_command}")
            return
        subprocess.run(["bash", "-lc", shell_command], check=True)
        return
    if args.env_command == "ensure":
        profile = normalize_env_profile(args.profile)
        env_script = ensure_env_tool(config)
        env_vars = os.environ.copy()
        env_vars.update(
            {
                "STEM_WORKSPACE_ROOT": str(config.workspace_root),
                "STEM_SOCKET_ROOT": str(config.socket_root),
                "STEM_PROFILE": profile,
            }
        )
        command = ["bash", str(env_script), "ensure", profile]
        if args.dry_run:
            print(f"profile\t{profile}")
            print(f"command\t{shlex.join(command)}")
            return
        subprocess.run(command, cwd=env_script.parent, env=env_vars, check=True)
        return
    raise SystemExit(f"Unsupported env command: {args.env_command}")


def run_probe(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    env_script = ensure_env_tool(config)
    command = ["bash", str(env_script), "probe", "list"]
    if args.json:
        command.append("--json")
    env = os.environ.copy()
    env.update(
        {
            "STEM_WORKSPACE_ROOT": str(config.workspace_root),
            "STEM_SOCKET_ROOT": str(config.socket_root),
            "STEM_PROFILE": "rp2350",
        }
    )
    subprocess.run(command, cwd=env_script.parent, env=env, check=True)


def run_mcp(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    """Manage the dynamic MCP target or serve one explicitly bound instance."""
    if args.mcp_kind == "list":
        print_mcp_instances(config)
        return
    if args.mcp_kind == "select":
        select_mcp_instance(config, args.selector)
        return
    if args.mcp_kind == "status":
        print_mcp_selection(config)
        return

    env_script = ensure_env_tool(config)
    wrapper = env_script.parent / "scripts" / f"mcp-{args.mcp_kind}.sh"
    if not wrapper.is_file():
        raise SystemExit(f"Missing MCP wrapper in environment sources: {wrapper}")
    env = os.environ.copy()
    env.update(
        {
            "STEM_WORKSPACE_ROOT": str(config.workspace_root),
            "STEM_SOCKET_ROOT": str(config.socket_root),
        }
    )
    os.execvpe(str(wrapper), [str(wrapper), args.instance], env)


def mcp_selection_root(config: WorkspaceConfig) -> Path:
    """Return the short, stable host path used by generic MCP providers."""
    return config.socket_root / "mcp-selected"


def mcp_registry_records(config: WorkspaceConfig) -> list[dict]:
    registry_root = config.socket_root / "registry"
    records: list[dict] = []
    for path in sorted(registry_root.glob("*.json")):
        try:
            record = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            continue
        if isinstance(record, dict):
            record["_registry_path"] = str(path)
            records.append(record)
    return records


def is_unix_socket(path: Path) -> bool:
    try:
        return stat.S_ISSOCK(path.stat().st_mode)
    except OSError:
        return False


def probe_mcp_socket_servers(record: dict, nvim_socket: Path, tmux_socket: Path) -> None:
    runtime = str(record["runtime"])
    container_name = str(record["container_name"])
    probes = (
        (
            "Neovim",
            [runtime, "exec", container_name, "nvim", "--server", str(nvim_socket), "--remote-expr", "1"],
        ),
        (
            "tmux",
            [runtime, "exec", container_name, "tmux", "-S", str(tmux_socket), "list-sessions"],
        ),
    )
    failures: list[str] = []
    for name, command in probes:
        try:
            result = subprocess.run(
                command,
                check=False,
                capture_output=True,
                text=True,
                timeout=3,
            )
        except subprocess.TimeoutExpired:
            failures.append(f"{name}: connection timed out")
            continue
        if result.returncode != 0:
            detail = (result.stderr or result.stdout).strip().splitlines()
            failures.append(f"{name}: {detail[-1] if detail else 'connection failed'}")
    if failures:
        raise SystemExit("MCP socket files exist, but their servers are not live:\n" + "\n".join(failures))


def validated_mcp_record(record: dict) -> tuple[dict, Path, Path]:
    required = (
        "runtime",
        "container_name",
        "container_id",
        "instance_key",
        "profile",
        "target",
        "socket_dir",
    )
    missing = [key for key in required if not str(record.get(key, "")).strip()]
    if missing:
        raise SystemExit(f"Incomplete MCP registry ({', '.join(missing)}): {record.get('_registry_path', '?')}")

    runtime = str(record["runtime"])
    container_name = str(record["container_name"])
    expected_id = str(record["container_id"])
    expected_instance_key = str(record["instance_key"])
    if shutil.which(runtime) is None:
        raise SystemExit(f"Recorded container runtime is unavailable: {runtime}")

    inspected = subprocess.run(
        [runtime, "inspect", container_name],
        check=False,
        capture_output=True,
        text=True,
    )
    if inspected.returncode != 0:
        raise SystemExit(f"Container is not available: {container_name}")
    try:
        inspection = json.loads(inspected.stdout)[0]
    except (IndexError, KeyError, TypeError, json.JSONDecodeError) as error:
        raise SystemExit(f"Cannot inspect container identity: {container_name}") from error
    actual_id = str(inspection.get("Id", ""))
    labels = inspection.get("Config", {}).get("Labels") or {}
    actual_instance_key = str(labels.get("edu.stem.instance-key", ""))
    actual_profile = str(labels.get("edu.stem.profile", ""))
    actual_target = str(labels.get("edu.stem.target", ""))
    if (
        actual_id != expected_id
        or actual_instance_key != expected_instance_key
        or actual_profile != str(record["profile"])
        or actual_target != str(record["target"])
    ):
        raise SystemExit(f"Stale MCP registry rejected for {container_name} (container identity changed).")

    socket_dir = Path(str(record["socket_dir"]))
    nvim_socket = socket_dir / "n.sock"
    tmux_socket = socket_dir / "t.sock"
    missing_sockets = [str(path) for path in (nvim_socket, tmux_socket) if not is_unix_socket(path)]
    if missing_sockets:
        raise SystemExit("MCP sockets are not live:\n" + "\n".join(missing_sockets))
    probe_mcp_socket_servers(record, nvim_socket, tmux_socket)
    return record, nvim_socket, tmux_socket


def resolve_mcp_record(config: WorkspaceConfig, selector: str) -> dict:
    selector = selector.strip()
    if not selector:
        raise SystemExit("MCP selector cannot be empty.")
    matches: list[dict] = []
    for record in mcp_registry_records(config):
        container_id = str(record.get("container_id", ""))
        if selector in {str(record.get("instance", "")), str(record.get("container_name", ""))}:
            matches.append(record)
        elif len(selector) >= 12 and container_id.startswith(selector):
            matches.append(record)
    unique = {str(record.get("container_id", "")): record for record in matches}
    if not unique:
        raise SystemExit(
            f"No registered MCP container matches {selector!r}. "
            "Use 'stemctl mcp list' to see available instances."
        )
    if len(unique) > 1:
        names = ", ".join(sorted(str(record.get("instance", "?")) for record in unique.values()))
        raise SystemExit(f"Ambiguous MCP selector {selector!r}: {names}")
    return next(iter(unique.values()))


def select_mcp_instance(config: WorkspaceConfig, selector: str) -> None:
    record, nvim_socket, tmux_socket = validated_mcp_record(resolve_mcp_record(config, selector))
    container_id = str(record["container_id"])
    cid12 = container_id[:12]
    selection_root = mcp_selection_root(config)
    target_root = selection_root / "targets" / cid12
    target_root.mkdir(parents=True, exist_ok=True)
    os.chmod(selection_root, 0o700)
    os.chmod(selection_root / "targets", 0o700)
    os.chmod(target_root, 0o700)

    for link, target in ((target_root / "n.sock", nvim_socket), (target_root / "t.sock", tmux_socket)):
        temporary = link.with_name(f".{link.name}.{os.getpid()}")
        temporary.unlink(missing_ok=True)
        temporary.symlink_to(target)
        os.replace(temporary, link)

    metadata = {
        key: record.get(key, "")
        for key in ("runtime", "container_name", "container_id", "profile", "target", "instance", "instance_key", "thread_key", "socket_dir")
    }
    metadata_path = target_root / "selection.json"
    temporary_metadata = metadata_path.with_name(f".{metadata_path.name}.{os.getpid()}")
    temporary_metadata.write_text(json.dumps(metadata, indent=2) + "\n", encoding="utf-8")
    os.replace(temporary_metadata, metadata_path)

    current = selection_root / "current"
    temporary_current = selection_root / f".current.{os.getpid()}"
    temporary_current.unlink(missing_ok=True)
    temporary_current.symlink_to(Path("targets") / cid12)
    os.replace(temporary_current, current)

    print(f"instance\t{record.get('instance', '')}")
    print(f"container\t{record.get('container_name', '')}")
    print(f"container_id\t{container_id}")
    print(f"nvim_socket\t{current / 'n.sock'}")
    print(f"tmux_socket\t{current / 't.sock'}")


def print_mcp_selection(config: WorkspaceConfig) -> None:
    current = mcp_selection_root(config) / "current"
    metadata_path = current / "selection.json"
    try:
        record = json.loads(metadata_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise SystemExit("No dynamic MCP container is selected. Use 'stemctl mcp select INSTANCE'.") from error
    record, _, _ = validated_mcp_record(record)
    print(f"instance\t{record.get('instance', '')}")
    print(f"container\t{record.get('container_name', '')}")
    print(f"container_id\t{record.get('container_id', '')}")
    print(f"nvim_socket\t{current / 'n.sock'}")
    print(f"tmux_socket\t{current / 't.sock'}")


def print_mcp_instances(config: WorkspaceConfig) -> None:
    print("container_id\tinstance\tprofile\tcontainer\tsockets")
    for record in sorted(mcp_registry_records(config), key=lambda item: str(item.get("instance", ""))):
        socket_dir = Path(str(record.get("socket_dir", "")))
        live = False
        if is_unix_socket(socket_dir / "n.sock") and is_unix_socket(socket_dir / "t.sock"):
            try:
                validated_mcp_record(record)
                live = True
            except SystemExit:
                pass
        print(
            f"{str(record.get('container_id', ''))[:12]}\t{record.get('instance', '')}\t"
            f"{record.get('profile', '')}\t{record.get('container_name', '')}\t"
            f"{'ready' if live else 'not-ready'}"
        )


def run_profile_action(config: WorkspaceConfig, args: argparse.Namespace, action: str) -> None:
    profile = normalize_env_profile(args.profile)
    selector_values, pane_target = split_pane_selector(args.selector, getattr(args, "pane", None))
    instance = getattr(args, "instance", None)
    instance_only_actions = {"status", "attach", "stop", "rm"}
    if instance and action in instance_only_actions and not selector_values:
        # A stable logical instance is sufficient for lifecycle operations on
        # an existing container.  Resolving workspace defaults here would make
        # the operation depend on the currently selected (or still checked
        # out) card, even though the environment launcher does not need it.
        series_name = card_name = task_name = None
    else:
        series_name, card_name, task_name = resolve_env_target(
            config, selector_values, getattr(args, "task", None)
        )
    shell_command, details = env_action_command(
        config,
        profile,
        action,
        series_name=series_name,
        card_name=card_name,
        task_name=task_name,
        instance=instance,
        editor=getattr(args, "editor", None),
        target=getattr(args, "target", None),
        device=getattr(args, "device", None),
        deploy_backend=getattr(args, "backend", None),
        allow_existing_firmware=getattr(args, "allow_existing_firmware", False),
    )
    if pane_target:
        pane_details = send_command_to_tmux_pane(shell_command, pane_target, dry_run=args.dry_run)
        details.update(pane_details)
        if args.dry_run:
            print_env_plan(details)
            return
        print_env_plan(details)
        return
    if args.dry_run:
        print_env_plan(details)
        return
    subprocess.run(["bash", "-lc", shell_command], check=True)


SESSION_STATUS_READY = "opracowane"
SESSION_STATUS_DRAFT = "robocze"
SESSION_STATUS_MISSING = "brak"


def ordered_series_names(config: WorkspaceConfig) -> list[str]:
    names = list(workspace_series_manifests(config))
    if names:
        return names
    for path in [*series_directories(config), *source_series_directories(config)]:
        if not path.name.startswith(".") and path.name not in names:
            names.append(path.name)
    return names


def resolve_session_series(config: WorkspaceConfig, selector: str | None) -> str:
    value = (selector or config.defaults.get("series") or "").strip()
    if not value:
        raise SystemExit("Missing series selector. Pass --series or configure defaults.series.")
    names = ordered_series_names(config)
    if value in names:
        return value
    if value.isdigit():
        index = int(value) - 1
        if 0 <= index < len(names):
            return names[index]
        raise SystemExit(f"Series number {value} is outside 1..{len(names)}.")
    matches = [name for name in names if value.lower() in name.lower()]
    if len(matches) == 1:
        return matches[0]
    if not matches:
        raise SystemExit(f"Missing series '{value}'. Available: {', '.join(names)}")
    raise SystemExit(f"Ambiguous series '{value}'. Matches: {', '.join(matches)}")


def read_card_source(card_path: Path) -> dict | None:
    source = card_path / "json" / "card_source.json"
    try:
        value = json.loads(source.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    return value if isinstance(value, dict) else None


def resolve_session_card(config: WorkspaceConfig, series_name: str, selector: str | None) -> str:
    value = (selector or default_card_for_series(config, series_name) or "").strip()
    if not value:
        raise SystemExit("Missing card selector. Pass --card or configure a default card.")
    cards = source_card_infos(config, series_name)
    if value.isdigit():
        number = int(value)
        index = number - 1
        if 0 <= index < len(cards):
            return cards[index].card_id
        raise SystemExit(f"Card number {value} is outside 1..{len(cards)} in series {series_name}.")
    return resolve_card_info(config, series_name, value).card_id


def task_number(value: str) -> int | None:
    match = re.search(r"(?i)(?:^|[^a-z0-9])task[-_]?0*(\d+)(?=$|[^0-9])", value)
    return int(match.group(1)) if match else None


def same_task(left: str, right: str) -> bool:
    if left.lower() == right.lower():
        return True
    left_number = task_number(left)
    right_number = task_number(right)
    return left_number is not None and left_number == right_number


def resolve_session_target(config: WorkspaceConfig, args: argparse.Namespace) -> SessionTarget:
    series_name = resolve_session_series(config, getattr(args, "series", None))
    card_name = resolve_session_card(config, series_name, getattr(args, "card", None))
    card_path = resolve_workspace_card_path(config, series_name, card_name)
    task_name = resolve_submission_task(config, card_path, getattr(args, "task", None))
    profile = normalize_env_profile(getattr(args, "profile", "hazard3-sim"))
    target_name = getattr(args, "target", None) or PROFILE_DEFAULT_TARGETS[profile]
    if target_name not in PROFILE_TARGETS[profile]:
        supported = ", ".join(sorted(PROFILE_TARGETS[profile]))
        raise SystemExit(
            f"Target {target_name!r} does not belong to profile {profile}. "
            f"Available targets: {supported}."
        )
    instance = getattr(args, "instance", None) or default_env_instance(
        profile, series_name, card_name, task_name, "session"
    )
    return SessionTarget(
        series=series_name,
        card=card_name,
        card_path=card_path,
        task=task_name,
        profile=profile,
        target=target_name,
        instance=instance,
    )


def card_navigation_catalog(card: dict | None) -> dict:
    entries: list[dict] = []
    if not card:
        return {"schema": "stem-card-navigation-catalog.v1", "levels": [], "entries": entries}
    task_ids: list[str] = []
    block_ids: dict[str, list[str]] = {}
    global_snapshots: list[str] = []
    for section_index, section in enumerate(card.get("sections", [])):
        if not isinstance(section, dict):
            continue
        for asset_index, asset in enumerate(section.get("assets", [])):
            interactive = asset.get("interactive") if isinstance(asset, dict) else None
            if not isinstance(interactive, dict) or not interactive.get("phases"):
                continue
            task_value = interactive.get("task") if isinstance(interactive.get("task"), dict) else {}
            block_value = interactive.get("block") if isinstance(interactive.get("block"), dict) else {}
            task_id = str(task_value.get("id") or f"section-{section_index + 1}")
            task_label = str(task_value.get("label") or section.get("title") or task_id)
            block_id = str(block_value.get("id") or str(asset.get("label", "")).removeprefix("fig:") or f"asset-{asset_index + 1}")
            block_label = str(block_value.get("label") or interactive.get("title") or asset.get("caption") or block_id)
            if task_id not in task_ids:
                task_ids.append(task_id)
            blocks = block_ids.setdefault(task_id, [])
            if block_id not in blocks:
                blocks.append(block_id)
            block_snapshots: list[str] = []
            for phase_index, phase in enumerate(interactive.get("phases", [])):
                if not isinstance(phase, dict):
                    continue
                phase_id = str(phase.get("id") or f"phase-{phase_index + 1}")
                phase_label = str(phase.get("label") or f"Faza {phase_index + 1}")
                phase_snapshots: list[str] = []
                for step_index, step in enumerate(phase.get("steps", [])):
                    if not isinstance(step, dict):
                        continue
                    step_id = str(step.get("id") or f"step-{step_index + 1}")
                    step_label = str(step.get("label") or f"Krok {step_index + 1}")
                    try:
                        step_number = int(step.get("number", step_index + 1))
                    except (TypeError, ValueError):
                        step_number = step_index + 1
                    snapshot_ref = str(step.get("snapshot_ref") or "") or None
                    if snapshot_ref and snapshot_ref not in phase_snapshots:
                        phase_snapshots.append(snapshot_ref)
                    if snapshot_ref and snapshot_ref not in block_snapshots:
                        block_snapshots.append(snapshot_ref)
                    if snapshot_ref and snapshot_ref not in global_snapshots:
                        global_snapshots.append(snapshot_ref)
                    entries.append(
                        {
                            "task": {"id": task_id, "label": task_label, "index": task_ids.index(task_id)},
                            "block": {"id": block_id, "label": block_label, "index": blocks.index(block_id)},
                            "phase": {"id": phase_id, "label": phase_label, "index": phase_index},
                            "step": {
                                "id": step_id,
                                "label": step_label,
                                "number": step_number,
                                "index": step_index,
                                "global_index": len(entries),
                            },
                            "snapshot": (
                                {
                                    "ref": snapshot_ref,
                                    "index": phase_snapshots.index(snapshot_ref),
                                    "block_index": block_snapshots.index(snapshot_ref),
                                    "global_index": global_snapshots.index(snapshot_ref),
                                }
                                if snapshot_ref
                                else None
                            ),
                            "asset": {
                                "label": str(asset.get("label") or ""),
                                "section_index": section_index,
                                "asset_index": asset_index,
                            },
                        }
                    )
    return {
        "schema": "stem-card-navigation-catalog.v1",
        "levels": ["task", "block", "phase", "step", "snapshot"],
        "entries": entries,
    }


def development_status(values: list[str]) -> str:
    if values and all(value == SESSION_STATUS_READY for value in values):
        return SESSION_STATUS_READY
    if values and any(value != SESSION_STATUS_MISSING for value in values):
        return SESSION_STATUS_DRAFT
    return SESSION_STATUS_MISSING


def checkpoint_items(card: dict | None) -> dict:
    value = card.get("debug_checkpoints", {}).get("items", {}) if card else {}
    return value if isinstance(value, dict) else {}


def valid_checkpoint_recipe(recipe: object) -> bool:
    if not isinstance(recipe, dict):
        return False
    stop = recipe.get("stop")
    if not isinstance(stop, dict) or not isinstance(stop.get("symbol"), str):
        return False
    if not isinstance(stop.get("offset"), int) or stop["offset"] < 0:
        return False
    verify = recipe.get("verify", {"expressions": []})
    return isinstance(verify, dict) and isinstance(verify.get("expressions", []), list)


def navigation_entry_status(entry: dict, recipes: dict) -> str:
    snapshot = entry.get("snapshot") or {}
    reference = snapshot.get("ref") if isinstance(snapshot, dict) else None
    if reference and valid_checkpoint_recipe(recipes.get(reference)):
        return SESSION_STATUS_READY
    if reference or entry.get("step"):
        return SESSION_STATUS_DRAFT
    return SESSION_STATUS_MISSING


def card_material_status(card_path: Path) -> str:
    source = read_card_source(card_path)
    if source is None:
        return SESSION_STATUS_MISSING if not card_path.is_dir() else SESSION_STATUS_DRAFT
    declared = str(source.get("card", {}).get("status", "")).strip().lower()
    if declared in {"gotowa", "gotowy", "ready", "opracowane", "opracowana", "published"}:
        return SESSION_STATUS_READY
    return SESSION_STATUS_DRAFT


def print_session_choices(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    series_names = ordered_series_names(config)
    raw_series = getattr(args, "series", None) or config.defaults.get("series")
    selected_series = resolve_session_series(config, raw_series) if raw_series else None
    raw_card = getattr(args, "card", None)
    if selected_series and raw_card is None:
        raw_card = default_card_for_series(config, selected_series)
    selected_card = resolve_session_card(config, selected_series, raw_card) if selected_series and raw_card else None
    cards = source_card_infos(config, selected_series) if selected_series else []
    card_info = resolve_card_info(config, selected_series, selected_card) if selected_series and selected_card else None
    card_path = workspace_card_path_for_info(config, card_info) if card_info else None
    source = read_card_source(card_path) if card_path else None
    catalog = card_navigation_catalog(source)
    recipes = checkpoint_items(source)
    explicit_task = getattr(args, "task", None)
    selected_task = None
    if explicit_task and card_path and card_path.is_dir():
        selected_task = resolve_submission_task(config, card_path, explicit_task)

    series_rows = [
        [str(index), "*" if name == selected_series else "", name]
        for index, name in enumerate(series_names, 1)
    ]
    card_rows = []
    for index, info in enumerate(cards, 1):
        path = workspace_card_path_for_info(config, info)
        card_rows.append(
            [
                str(index),
                "*" if info.card_id == selected_card else "",
                info.card_id,
                info.title or info.repo,
                card_material_status(path),
            ]
        )

    nav_status_by_task: dict[str, list[str]] = {}
    for entry in catalog.get("entries", []):
        nav_status_by_task.setdefault(entry["task"]["id"], []).append(navigation_entry_status(entry, recipes))
    task_rows = []
    if card_path and card_path.is_dir():
        for entry in task_entries(card_path):
            matching = [
                status
                for task_id, statuses in nav_status_by_task.items()
                if same_task(task_id, entry.name)
                for status in statuses
            ]
            status = development_status(matching) if matching else SESSION_STATUS_DRAFT
            task_rows.append(
                [
                    str(task_number(entry.name) or len(task_rows) + 1),
                    "*" if selected_task == entry.name else "",
                    entry.name,
                    status,
                    str(entry.path),
                ]
            )

    uml_rows = []
    for entry in catalog.get("entries", []):
        if selected_task and not same_task(entry["task"]["id"], selected_task):
            continue
        snapshot = entry.get("snapshot") or {}
        uml_rows.append(
            [
                entry["task"]["id"],
                str(entry["block"]["index"] + 1),
                str(entry["phase"]["index"] + 1),
                str(entry["step"]["number"]),
                str(snapshot.get("index", -1) + 1) if snapshot else "-",
                str(snapshot.get("block_index", -1) + 1) if snapshot else "-",
                str(snapshot.get("global_index", -1) + 1) if snapshot else "-",
                entry["phase"]["id"],
                entry["step"]["id"],
                str(snapshot.get("ref") or "-"),
                navigation_entry_status(entry, recipes),
                "/".join(
                    (
                        entry["task"]["id"],
                        entry["block"]["id"],
                        entry["phase"]["id"],
                        entry["step"]["id"],
                    )
                ),
                entry["step"]["label"],
            ]
        )

    if getattr(args, "json", False):
        print(
            json.dumps(
                {
                    "schema": "stem-session-choices.v1",
                    "series": [
                        {"number": int(row[0]), "selected": bool(row[1]), "id": row[2]}
                        for row in series_rows
                    ],
                    "cards": [
                        {
                            "number": int(row[0]),
                            "selected": bool(row[1]),
                            "id": row[2],
                            "title": row[3],
                            "status": row[4],
                        }
                        for row in card_rows
                    ],
                    "tasks": [
                        {
                            "number": int(row[0]),
                            "selected": bool(row[1]),
                            "id": row[2],
                            "status": row[3],
                            "source": row[4],
                        }
                        for row in task_rows
                    ],
                    "uml": [
                        {
                            "task_id": row[0],
                            "block_number": int(row[1]),
                            "phase_number": int(row[2]),
                            "step_number": int(row[3]),
                            "snapshot_phase_number": None if row[4] == "-" else int(row[4]),
                            "snapshot_block_number": None if row[5] == "-" else int(row[5]),
                            "snapshot_global_number": None if row[6] == "-" else int(row[6]),
                            "phase_id": row[7],
                            "step_id": row[8],
                            "snapshot_ref": None if row[9] == "-" else row[9],
                            "status": row[10],
                            "selector": row[11],
                            "label": row[12],
                        }
                        for row in uml_rows
                    ],
                    "status_values": [SESSION_STATUS_READY, SESSION_STATUS_DRAFT, SESSION_STATUS_MISSING],
                },
                indent=2,
                ensure_ascii=False,
            )
        )
        return
    print("Serie:")
    print_table(["nr", "wybrana", "seria"], series_rows)
    print("\nKarty:")
    print_table(["nr", "wybrana", "karta", "tytuł", "status"], card_rows)
    print("\nTaski:")
    print_table(["nr", "wybrany", "task", "status", "źródło"], task_rows)
    print("\nEtapy UML:")
    print_table(
        [
            "task",
            "blok",
            "faza",
            "krok",
            "snap/faza",
            "snap/blok",
            "snap/global",
            "phase_id",
            "step_id",
            "snapshot",
            "status",
            "selector",
            "opis",
        ],
        uml_rows,
    )


def session_stage_selectors(
    args: argparse.Namespace,
    task_name: str | None = None,
) -> tuple[str | None, str | None, str | None, str | None]:
    block = getattr(args, "block", None)
    phase = getattr(args, "phase", None)
    step = getattr(args, "step", None)
    snapshot = getattr(args, "snapshot", None)
    stage = (getattr(args, "stage", None) or "").strip()
    first = bool(getattr(args, "first", False))
    if first and (stage or any(value for value in (block, phase, step, snapshot))):
        raise SystemExit("--first cannot be combined with another UML stage selector.")
    if not stage:
        return block, phase, step, snapshot
    if any(value for value in (block, phase, step, snapshot)):
        raise SystemExit("Pass either --stage or the separate --block/--phase/--step/--snapshot selectors.")
    if re.fullmatch(r"(?i)task\d+\.[a-z0-9.-]+", stage):
        return None, None, None, stage
    parts = [part.strip() for part in re.split(r"[/:]", stage) if part.strip()]
    if len(parts) == 1:
        return None, None, parts[0], None
    if len(parts) == 2:
        return None, parts[0], parts[1], None
    if len(parts) == 3:
        return parts[0], parts[1], parts[2], None
    if len(parts) == 4:
        if task_name and not same_task(parts[0], task_name):
            raise SystemExit(f"Stage selector belongs to {parts[0]}, selected task is {task_name}.")
        return parts[1], parts[2], parts[3], None
    raise SystemExit(
        "--stage accepts STEP, PHASE/STEP, BLOCK/PHASE/STEP, "
        "TASK/BLOCK/PHASE/STEP or a snapshot_ref."
    )


def navigation_level_path(entry: dict, level: str) -> str:
    task_id = str(entry.get("task", {}).get("id", ""))
    block_id = str(entry.get("block", {}).get("id", ""))
    phase_id = str(entry.get("phase", {}).get("id", ""))
    if level == "block":
        return f"{task_id}/{block_id}"
    if level == "phase":
        return f"{task_id}/{block_id}/{phase_id}"
    if level == "snapshot":
        return f"{task_id}/{block_id}/{entry.get('snapshot', {}).get('ref', '')}"
    return ""


def unique_navigation_values(entries: list[dict], level: str) -> list[tuple[str, dict]]:
    values: list[tuple[str, dict]] = []
    seen: set[str] = set()
    for entry in entries:
        value = entry.get(level)
        if not isinstance(value, dict):
            continue
        key = navigation_level_path(entry, level)
        if not key or key == "None" or key in seen:
            continue
        seen.add(key)
        values.append((key, value))
    return values


def filter_navigation_level(entries: list[dict], level: str, selector: str | None) -> list[dict]:
    if selector is None:
        return entries
    values = unique_navigation_values(entries, level)
    selected: tuple[str, dict] | None = None
    raw = selector.strip()
    if raw.isdigit():
        if level in {"phase", "snapshot"}:
            parent_paths = {
                "/".join(path.split("/")[:2])
                for path, _ in values
            }
            if len(parent_paths) > 1:
                raise SystemExit(f"Numeric {level} requires an explicit --block parent.")
        index = int(raw) - 1
        if 0 <= index < len(values):
            selected = values[index]
    else:
        key_name = "ref" if level == "snapshot" else "id"
        exact = [item for item in values if str(item[1].get(key_name, "")).lower() == raw.lower()]
        matches = exact or [
            item
            for item in values
            if raw.lower() in str(item[1].get(key_name, "")).lower()
            or raw.lower() in str(item[1].get("label", "")).lower()
        ]
        if len(matches) == 1:
            selected = matches[0]
        elif len(matches) > 1:
            choices = ", ".join(path for path, _ in matches)
            raise SystemExit(f"Ambiguous {level} selector '{raw}'. Matches: {choices}")
    if selected is None:
        choices = ", ".join(path for path, _ in values)
        raise SystemExit(f"Missing {level} '{raw}'. Available: {choices}")
    selected_path = selected[0]
    return [
        entry
        for entry in entries
        if isinstance(entry.get(level), dict) and navigation_level_path(entry, level) == selected_path
    ]


def resolve_navigation_entry(catalog: dict, task_name: str, args: argparse.Namespace) -> dict:
    entries = [
        entry
        for entry in catalog.get("entries", [])
        if isinstance(entry, dict) and same_task(str(entry.get("task", {}).get("id", "")), task_name)
    ]
    if not entries:
        raise SystemExit(f"Card server has no UML navigation entries for {task_name}.")
    block, phase, step, snapshot = session_stage_selectors(args, task_name)
    entries = filter_navigation_level(entries, "block", block)
    entries = filter_navigation_level(entries, "phase", phase)
    entries = filter_navigation_level(entries, "snapshot", snapshot)
    if step is not None:
        raw = step.strip()
        matches: list[dict] = []
        if raw.isdigit():
            number = int(raw)
            matches = [entry for entry in entries if int(entry.get("step", {}).get("number", -1)) == number]
            if not matches and phase is not None:
                matches = [entry for entry in entries if int(entry.get("step", {}).get("index", -1)) == number - 1]
        else:
            exact = [entry for entry in entries if str(entry.get("step", {}).get("id", "")).lower() == raw.lower()]
            matches = exact or [
                entry
                for entry in entries
                if raw.lower() in str(entry.get("step", {}).get("id", "")).lower()
                or raw.lower() in str(entry.get("step", {}).get("label", "")).lower()
            ]
        if not matches:
            choices = ", ".join(
                f"{entry.get('step', {}).get('number')}:{entry.get('step', {}).get('id')}" for entry in entries
            )
            raise SystemExit(f"Missing step '{raw}'. Available: {choices}")
        if len(matches) > 1:
            raise SystemExit(f"Ambiguous step '{raw}'. Add --phase or use a step ID.")
        entries = [matches[0]]
    if not entries:
        raise SystemExit("The UML selectors do not identify any stage.")
    return entries[0]


def card_api_json(
    base_url: str,
    path: str,
    method: str = "GET",
    payload: dict | None = None,
    timeout: float = 30,
) -> dict:
    data = None
    headers = {"accept": "application/json"}
    if payload is not None:
        data = json.dumps(payload).encode("utf-8")
        headers["content-type"] = "application/json"
    request = urlrequest.Request(
        f"{base_url.rstrip('/')}{path}", data=data, headers=headers, method=method
    )
    try:
        with urlrequest.urlopen(request, timeout=timeout) as response:
            value = json.loads(response.read().decode("utf-8"))
    except urlerror.HTTPError as error:
        detail = error.read().decode("utf-8", errors="replace")
        raise SystemExit(f"Card API {method} {path} returned HTTP {error.code}: {detail}") from error
    except (urlerror.URLError, TimeoutError, json.JSONDecodeError) as error:
        raise SystemExit(
            f"Card API is unavailable at {base_url}: {error}. Start the selected card server or pass --card-url."
        ) from error
    if not isinstance(value, dict):
        raise SystemExit(f"Card API {path} returned a non-object JSON response.")
    return value


def session_stage_requested(args: argparse.Namespace) -> bool:
    return bool(getattr(args, "first", False)) or any(
        getattr(args, name, None)
        for name in ("stage", "block", "phase", "step", "snapshot")
    )


def verify_card_api_identity(base_url: str, session: SessionTarget, timeout: float) -> dict:
    local_source = read_card_source(session.card_path)
    card_file = session.card_path / "json" / "card_source.json"
    if not local_source or not card_file.is_file():
        raise SystemExit(f"Cannot verify card API identity without {card_file}.")
    expected = {
        "id": str(local_source.get("card", {}).get("id", "")),
        "uuid": str(local_source.get("card", {}).get("uuid", "")),
        "version": str(local_source.get("card", {}).get("version", "")),
        "source_sha256": hashlib.sha256(card_file.read_bytes()).hexdigest(),
    }
    if not all(expected.values()):
        raise SystemExit(f"Card identity is incomplete in {card_file}.")
    served = card_api_json(base_url, "/api/identity", timeout=timeout)
    mismatches = [name for name, value in expected.items() if str(served.get(name, "")) != value]
    if mismatches:
        raise SystemExit(
            f"Card API mismatch for {', '.join(mismatches)} at {base_url}. "
            "Start the correct card server or pass --card-url."
        )
    return served


def validate_session_checkpoint(session: SessionTarget, snapshot_ref: str) -> None:
    source = read_card_source(session.card_path)
    registry = source.get("debug_checkpoints") if source else None
    if not isinstance(registry, dict):
        raise SystemExit("The selected card has no debug_checkpoints registry.")
    targets = registry.get("targets")
    if not isinstance(targets, dict) or not isinstance(targets.get(session.profile), dict):
        raise SystemExit(f"The card has no checkpoint adapter for profile {session.profile}.")
    artifact = registry.get("artifact")
    if not isinstance(artifact, dict) or not artifact.get("source") or not artifact.get("elf"):
        raise SystemExit("The checkpoint artifact identity is incomplete.")
    recipe = checkpoint_items(source).get(snapshot_ref)
    if not valid_checkpoint_recipe(recipe):
        raise SystemExit(f"Snapshot {snapshot_ref} has no complete checkpoint recipe.")


def prepare_session_stage(session: SessionTarget, args: argparse.Namespace, online: bool) -> dict:
    verify_card_api_identity(args.card_url, session, args.timeout)
    catalog = card_api_json(args.card_url, "/api/navigation/catalog", timeout=args.timeout)
    entry = resolve_navigation_entry(catalog, session.task, args)
    snapshot = entry.get("snapshot") or {}
    snapshot_ref = snapshot.get("ref") if isinstance(snapshot, dict) else None
    if online:
        if not snapshot_ref:
            raise SystemExit("The selected UML step has no snapshot and cannot be replayed online.")
        validate_session_checkpoint(session, str(snapshot_ref))
    return entry


def select_session_stage(
    config: WorkspaceConfig,
    session: SessionTarget,
    args: argparse.Namespace,
    online: bool,
    prepared_entry: dict | None = None,
) -> dict:
    base_url = args.card_url
    entry = prepared_entry or prepare_session_stage(session, args, online)
    snapshot = entry.get("snapshot") or {}
    snapshot_ref = snapshot.get("ref") if isinstance(snapshot, dict) else None
    result: dict = {}
    if online:
        if not snapshot_ref:
            raise SystemExit("The selected UML step has no snapshot and cannot be replayed online.")
        expected_record = resolve_mcp_record(config, session.instance)
        validated_mcp_record(expected_record)
        validate_mcp_card_mount(expected_record, session.card_path)
        if str(expected_record.get("profile", "")) != session.profile:
            raise SystemExit("MCP profile changed after preflight.")
        if str(expected_record.get("target", "")) != session.target:
            raise SystemExit("MCP target changed after preflight.")
        select_mcp_instance(config, session.instance)
        expected_identity = verify_card_api_identity(base_url, session, args.timeout)
        result = card_api_json(
            base_url,
            "/api/checkpoint/activate",
            method="POST",
            payload={
                "snapshot_ref": snapshot_ref,
                "generation": int(time.time() * 1000),
                "expected_binding": {
                    key: str(expected_record.get(key, ""))
                    for key in ("container_id", "instance", "profile", "target")
                },
                "expected_identity": {
                    key: str(expected_identity.get(key, ""))
                    for key in ("id", "uuid", "version", "source_sha256")
                },
            },
            timeout=max(args.timeout, 25),
        )
        if result.get("status") != "ready":
            raise SystemExit(
                f"Checkpoint {snapshot_ref} did not reach ready: "
                f"{result.get('status', 'unknown')} — {result.get('message', 'no message')}"
            )
        verify_card_api_identity(base_url, session, args.timeout)
    navigation = card_api_json(
        base_url,
        "/api/navigation/current",
        method="PUT",
        payload={
            "task_id": entry["task"]["id"],
            "block_id": entry["block"]["id"],
            "phase_id": entry["phase"]["id"],
            "step_id": entry["step"]["id"],
            "snapshot_ref": snapshot_ref,
            "focus_level": args.focus,
            "sync_requested": online,
            "actor": "stemctl",
        },
        timeout=args.timeout,
    )
    card_api_json(
        base_url,
        "/api/viewer/state",
        method="PUT",
        payload={
            "actor": "stemctl",
            "nvim": {
                "visible": True,
                "sync": online,
                "control": online and args.control,
            },
        },
        timeout=args.timeout,
    )
    print(f"task\t{entry['task']['id']}")
    print(f"block\t{entry['block']['id']}")
    print(f"phase\t{entry['phase']['id']}")
    print(f"step\t{entry['step']['number']}:{entry['step']['id']}")
    print(f"snapshot\t{snapshot_ref or ''}")
    print(f"mode\t{'online' if online else 'offline'}")
    if not online:
        return navigation
    print(f"checkpoint_status\t{result.get('status', '')}")
    print(f"checkpoint_message\t{result.get('message', '')}")
    if result.get("actual_pc"):
        print(f"pc\t{result['actual_pc']}")
    return result


def mark_invoking_codex_pane() -> None:
    current_pane = os.environ.get("TMUX_PANE", "").strip()
    thread_id = os.environ.get("CODEX_THREAD_ID", "").strip()
    if not current_pane or not thread_id:
        return
    pane = subprocess.run(
        ["tmux", "display-message", "-p", "-t", current_pane, "#{pane_pid}"],
        check=False,
        capture_output=True,
        text=True,
    )
    if pane.returncode != 0 or not pane.stdout.strip().isdigit():
        return
    subprocess.run(
        ["tmux", "set-option", "-p", "-t", current_pane, "@stem_codex_pid", pane.stdout.strip()],
        check=False,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    subprocess.run(
        ["tmux", "set-option", "-p", "-t", current_pane, "@stem_codex_thread", thread_id],
        check=False,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )


def resolve_tmux_session_pane(raw_target: str, force: bool = False) -> tuple[str, str, str]:
    mark_invoking_codex_pane()
    target = raw_target.strip()
    current_match = re.fullmatch(r"current:(\d+)", target)
    if current_match:
        current_pane = os.environ.get("TMUX_PANE", "").strip()
        if not current_pane:
            raise SystemExit("--pane current:N requires stemctl to run inside tmux; pass an explicit target instead.")
        window = subprocess.run(
            ["tmux", "display-message", "-p", "-t", current_pane, "#{session_name}:#{window_index}"],
            check=False,
            capture_output=True,
            text=True,
        )
        if window.returncode != 0:
            raise SystemExit(f"Cannot resolve the current tmux window: {window.stderr.strip()}")
        target = f"{window.stdout.strip()}.{current_match.group(1)}"
    else:
        target = tmux_pane_target(target)
    resolved = subprocess.run(
        [
            "tmux",
            "display-message",
            "-p",
            "-t",
            target,
            "#{session_name}:#{window_index}.#{pane_index}\t#{pane_id}\t#{pane_current_path}"
            "\t#{pane_pid}\t#{@stem_codex_pid}\t#{@stem_codex_thread}",
        ],
        check=False,
        capture_output=True,
        text=True,
    )
    if resolved.returncode != 0:
        raise SystemExit(f"Missing tmux pane target '{raw_target}' ({target}): {resolved.stderr.strip()}")
    name, pane_id, current_path, pane_pid, protected_pid, protected_thread = resolved.stdout.rstrip("\n").split("\t", 5)
    current_pane = os.environ.get("TMUX_PANE", "").strip()
    if current_pane and pane_id == current_pane:
        raise SystemExit(
            f"Refusing to replace the pane running stemctl/Codex ({pane_id}). Choose another pane, normally current:0."
        )
    if protected_thread and protected_pid == pane_pid and not force:
        raise SystemExit(
            f"Refusing to replace pane {pane_id}: it is owned by Codex thread {protected_thread}. "
            "Use another pane or pass --force-pane deliberately."
        )
    return name, pane_id, current_path


def respawn_tmux_session_pane(
    raw_target: str,
    cwd: Path,
    shell_command: str | None,
    dry_run: bool = False,
    force: bool = False,
    stem_instance: str | None = None,
) -> dict[str, str]:
    name, pane_id, _ = resolve_tmux_session_pane(raw_target, force=force)
    shell = os.environ.get("SHELL", "/bin/bash")
    if shell_command:
        inner = f"{shell_command}; exec {shlex.quote(shell)} -l"
        pane_command = f"{shlex.quote(shell)} -lc {shlex.quote(inner)}"
    else:
        pane_command = f"exec {shlex.quote(shell)} -l"
    command = ["tmux", "respawn-pane", "-k", "-t", pane_id, "-c", str(cwd), pane_command]
    details = {
        "pane": raw_target,
        "tmux_target": name,
        "tmux_pane_id": pane_id,
        "tmux_command": shlex.join(command),
    }
    if not dry_run:
        subprocess.run(command, check=True)
        # A debugger session contains an inner tmux and Neovim.  If its host
        # pane remains in a split, the inner client inherits only that short
        # height and leaves an unusable empty area below the editor.  Zoom the
        # replacement pane after every attach/refresh/reset so it receives the
        # full host window height.  A later user Ctrl-a z simply restores the
        # ordinary host-pane layout.
        zoom_state = subprocess.run(
            ["tmux", "display-message", "-p", "-t", pane_id, "#{window_zoomed_flag}"],
            check=False,
            text=True,
            capture_output=True,
        )
        if zoom_state.returncode == 0 and zoom_state.stdout.strip() != "1":
            subprocess.run(
                ["tmux", "resize-pane", "-Z", "-t", pane_id],
                check=False,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
        subprocess.run(
            ["tmux", "set-option", "-p", "-t", pane_id, "@stem_role", "debugger" if shell_command else "shell"],
            check=False,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        instance_option = ["tmux", "set-option", "-p", "-t", pane_id]
        if shell_command and stem_instance:
            instance_option.extend(["@stem_instance", stem_instance])
        else:
            instance_option.extend(["-u", "@stem_instance"])
        subprocess.run(
            instance_option,
            check=False,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
    return details


def validate_mcp_card_mount(record: dict, card_path: Path) -> None:
    runtime = str(record.get("runtime", ""))
    container_name = str(record.get("container_name", ""))
    inspected = subprocess.run(
        [runtime, "inspect", container_name],
        check=False,
        capture_output=True,
        text=True,
    )
    if inspected.returncode != 0:
        raise SystemExit(f"Cannot inspect MCP container mount: {container_name}")
    try:
        inspection = json.loads(inspected.stdout)[0]
    except (IndexError, TypeError, json.JSONDecodeError) as error:
        raise SystemExit(f"Cannot decode MCP container mount: {container_name}") from error
    workspace_mounts = [
        mount
        for mount in inspection.get("Mounts", [])
        if isinstance(mount, dict) and mount.get("Destination") == "/workspace"
    ]
    if len(workspace_mounts) != 1:
        raise SystemExit(f"MCP container {container_name} has no unique /workspace bind mount.")
    source = Path(str(workspace_mounts[0].get("Source", ""))).resolve()
    if source != card_path.resolve():
        raise SystemExit(
            f"MCP container/card mismatch: {container_name} mounts {source}, selected card is {card_path.resolve()}."
        )


def ready_mcp_record(
    config: WorkspaceConfig,
    instance: str,
    card_path: Path | None = None,
    profile: str | None = None,
    target: str | None = None,
) -> tuple[dict | None, str]:
    try:
        record = resolve_mcp_record(config, instance)
        validated_mcp_record(record)
        if card_path is not None:
            validate_mcp_card_mount(record, card_path)
        if profile is not None and str(record.get("profile", "")) != profile:
            raise SystemExit(
                f"MCP profile mismatch: {record.get('profile', '<unknown>')} != {profile}."
            )
        if target is not None and str(record.get("target", "")) != target:
            raise SystemExit(
                f"MCP target mismatch: {record.get('target', '<unknown>')} != {target}."
            )
        return record, ""
    except SystemExit as error:
        return None, str(error)


def wait_for_session_mcp(
    config: WorkspaceConfig,
    instance: str,
    timeout: float,
    card_path: Path | None = None,
    profile: str | None = None,
    target: str | None = None,
) -> dict:
    deadline = time.monotonic() + timeout
    last_error = ""
    print(f"waiting_for_mcp\t{instance}", file=sys.stderr, flush=True)
    while time.monotonic() < deadline:
        record, last_error = ready_mcp_record(config, instance, card_path, profile, target)
        if record is not None:
            return record
        time.sleep(0.25)
    raise SystemExit(f"MCP for {instance} did not become ready within {timeout:g}s: {last_error}")


def session_env_command(config: WorkspaceConfig, session: SessionTarget, action: str) -> tuple[str, dict[str, str]]:
    return env_action_command(
        config,
        session.profile,
        action,
        series_name=session.series,
        card_name=session.card,
        task_name=session.task,
        instance=session.instance,
        editor="nvim",
        target=session.target,
    )


def print_session_target(session: SessionTarget) -> None:
    print(f"series\t{session.series}")
    print(f"card\t{session.card}")
    print(f"card_path\t{session.card_path}")
    print(f"task\t{session.task}")
    print(f"profile\t{session.profile}")
    print(f"target\t{session.target}")
    print(f"instance\t{session.instance}")


def session_status(config: WorkspaceConfig, session: SessionTarget, args: argparse.Namespace) -> None:
    data: dict = {
        "schema": "stem-session-status.v1",
        "series": session.series,
        "card": session.card,
        "card_path": str(session.card_path),
        "task": session.task,
        "profile": session.profile,
        "target": session.target,
        "instance": session.instance,
    }
    record, error = ready_mcp_record(
        config,
        session.instance,
        session.card_path,
        session.profile,
        session.target,
    )
    if record is None:
        data["mcp"] = {"status": "not-ready", "detail": error}
    else:
        data["mcp"] = {
            "status": "ready",
            "container": record.get("container_name", ""),
            "container_id": record.get("container_id", ""),
        }
    try:
        identity = verify_card_api_identity(args.card_url, session, min(args.timeout, 3))
        state = card_api_json(args.card_url, "/api/state", timeout=min(args.timeout, 3))
    except SystemExit as api_error:
        data["card_api"] = {"status": "offline", "detail": str(api_error)}
    else:
        navigation = state.get("navigation") or {}
        data["card_api"] = {
            "status": "online",
            "identity": identity,
            "phase": navigation.get("phase", {}).get("id", ""),
            "step": navigation.get("step", {}),
            "snapshot": navigation.get("snapshot_ref", ""),
            "sync": bool(navigation.get("sync_requested")),
        }
    if args.json:
        print(json.dumps(data, indent=2, ensure_ascii=False))
    else:
        for key in ("series", "card", "card_path", "task", "profile", "target", "instance"):
            print(f"{key}\t{data[key]}")
        print(f"mcp\t{data['mcp']['status']}")
        if data["mcp"]["status"] == "ready":
            print(f"container\t{data['mcp']['container']}")
            print(f"container_id\t{data['mcp']['container_id']}")
        else:
            print(f"mcp_detail\t{data['mcp']['detail']}")
        print(f"card_api\t{data['card_api']['status']}")
        if data["card_api"]["status"] == "online":
            api = data["card_api"]
            print(f"card_api_id\t{api['identity'].get('id', '')}")
            print(f"phase\t{api['phase']}")
            print(f"step\t{api['step'].get('number', '')}:{api['step'].get('id', '')}")
            print(f"snapshot\t{api['snapshot']}")
            print(f"sync\t{str(api['sync']).lower()}")
        else:
            print(f"card_api_detail\t{data['card_api']['detail']}")
    if args.strict and (
        data["mcp"]["status"] != "ready" or data["card_api"]["status"] != "online"
    ):
        raise SystemExit("Session is not fully ready.")


def validate_session_command_args(args: argparse.Namespace) -> None:
    command = args.session_command
    has_stage = session_stage_requested(args)
    if args.offline and args.control:
        raise SystemExit("--control requires online SYNC and cannot be combined with --offline.")
    if (args.offline or args.control) and not has_stage:
        raise SystemExit("--offline/--control require an explicit UML stage selector.")
    if command in {"stage", "checkpoint"} and not has_stage:
        raise SystemExit("stage/checkpoint requires a UML selector or explicit --first.")
    if command in {"status", "detach"} and has_stage:
        raise SystemExit(f"{command} does not accept UML stage selectors.")
    if command == "status" and args.dry_run:
        raise SystemExit("status is already read-only; --dry-run is not applicable.")
    if command in {"reset", "start"} and args.pane == "none" and has_stage and not args.offline:
        raise SystemExit(
            f"Online UML replay during {command} requires a debugger pane; "
            "omit --pane none or use --offline."
        )
    if command in {"attach", "detach"} and args.pane == "none":
        raise SystemExit(f"{command} requires --pane; use current:0 or an explicit tmux pane ID.")


def print_prepared_stage(entry: dict, online: bool) -> None:
    snapshot = entry.get("snapshot") or {}
    print(f"planned_task\t{entry.get('task', {}).get('id', '')}")
    print(f"planned_block\t{entry.get('block', {}).get('id', '')}")
    print(f"planned_phase\t{entry.get('phase', {}).get('id', '')}")
    print(
        f"planned_step\t{entry.get('step', {}).get('number', '')}:"
        f"{entry.get('step', {}).get('id', '')}"
    )
    print(f"planned_snapshot\t{snapshot.get('ref', '')}")
    print(f"planned_mode\t{'online' if online else 'offline'}")


def run_session(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    command = args.session_command
    if command == "choices":
        print_session_choices(config, args)
        return
    validate_session_command_args(args)
    session = resolve_session_target(config, args)
    if command == "status":
        session_status(config, session, args)
        return
    online = not args.offline
    prepared_stage = prepare_session_stage(session, args, online) if session_stage_requested(args) else None
    if command in {"reset", "start", "refresh", "attach", "detach"} and args.pane != "none":
        # Resolve and protect the target before rm/start/MCP selection. The
        # respawn call resolves it once more immediately before replacement.
        resolve_tmux_session_pane(args.pane, force=args.force_pane)
    if command == "refresh" and args.pane == "none" and prepared_stage is not None and online:
        record, error = ready_mcp_record(
            config,
            session.instance,
            session.card_path,
            session.profile,
            session.target,
        )
        if record is None:
            raise SystemExit(
                "Online refresh with --pane none requires an already live debugger: " + error
            )
    if command == "detach":
        details = respawn_tmux_session_pane(
            args.pane,
            session.card_path,
            None,
            dry_run=args.dry_run,
            force=args.force_pane,
            stem_instance=session.instance,
        )
        print_session_target(session)
        print_key_values(details)
        print("status\tdetached" if not args.dry_run else "status\tdry-run")
        return
    if command in {"stage", "checkpoint"}:
        print_session_target(session)
        print_prepared_stage(prepared_stage or {}, online)
        if args.dry_run:
            print("status\tdry-run")
            return
        if online:
            wait_for_session_mcp(
                config, session.instance, args.timeout, session.card_path, session.profile, session.target
            )
        select_session_stage(
            config,
            session,
            args,
            online=online,
            prepared_entry=prepared_stage,
        )
        return

    print_session_target(session)
    if prepared_stage is not None:
        print_prepared_stage(prepared_stage, online)
    if command == "reset":
        remove_command, remove_details = session_env_command(config, session, "rm")
        print(f"remove_command\t{remove_command}")
        if not args.dry_run:
            subprocess.run(["bash", "-lc", remove_command], check=True)
        if args.pane == "none":
            start_command, _ = session_env_command(config, session, "start")
            print(f"start_command\t{start_command}")
            if not args.dry_run:
                subprocess.run(["bash", "-lc", start_command], check=True)
            print_env_plan(remove_details)
            if prepared_stage is not None and not args.dry_run:
                select_session_stage(
                    config, session, args, online=False, prepared_entry=prepared_stage
                )
            return
        debug_command, debug_details = session_env_command(config, session, "debug")
        pane_details = respawn_tmux_session_pane(
            args.pane,
            session.card_path,
            debug_command,
            dry_run=args.dry_run,
            force=args.force_pane,
            stem_instance=session.instance,
        )
        print_key_values(pane_details)
        if args.dry_run:
            print_env_plan(debug_details)
            return
        wait_for_session_mcp(
            config, session.instance, args.timeout, session.card_path, session.profile, session.target
        )
        select_mcp_instance(config, session.instance)
        if prepared_stage is not None:
            select_session_stage(
                config, session, args, online=online, prepared_entry=prepared_stage
            )
        return

    if command == "start":
        if args.pane == "none":
            start_command, details = session_env_command(config, session, "start")
            if args.dry_run:
                print_env_plan(details)
            else:
                subprocess.run(["bash", "-lc", start_command], check=True)
                if prepared_stage is not None:
                    select_session_stage(
                        config, session, args, online=False, prepared_entry=prepared_stage
                    )
            return
        debug_command, details = session_env_command(config, session, "debug")
        pane_details = respawn_tmux_session_pane(
            args.pane,
            session.card_path,
            debug_command,
            dry_run=args.dry_run,
            force=args.force_pane,
            stem_instance=session.instance,
        )
        print_key_values(pane_details)
        if args.dry_run:
            print_env_plan(details)
            return
        wait_for_session_mcp(
            config, session.instance, args.timeout, session.card_path, session.profile, session.target
        )
        select_mcp_instance(config, session.instance)
        if prepared_stage is not None:
            select_session_stage(
                config, session, args, online=online, prepared_entry=prepared_stage
            )
        return

    if command == "refresh":
        start_command, details = session_env_command(config, session, "start")
        if args.dry_run:
            print_env_plan(details)
        else:
            subprocess.run(["bash", "-lc", start_command], check=True)
        record, _ = (
            ready_mcp_record(
                config,
                session.instance,
                session.card_path,
                session.profile,
                session.target,
            )
            if not args.dry_run
            else (None, "")
        )
        pane_action = "attach" if record is not None else "debug"
        pane_command, pane_env_details = session_env_command(config, session, pane_action)
        if args.pane != "none":
            pane_details = respawn_tmux_session_pane(
                args.pane,
                session.card_path,
                pane_command,
                dry_run=args.dry_run,
                force=args.force_pane,
                stem_instance=session.instance,
            )
            print_key_values(pane_details)
        elif record is None and not args.dry_run and (prepared_stage is None or online):
            raise SystemExit("The debugger is not ready and refresh cannot create it with --pane none.")
        if args.dry_run:
            print_env_plan(pane_env_details)
            return
        if record is not None or args.pane != "none":
            wait_for_session_mcp(
                config, session.instance, args.timeout, session.card_path, session.profile, session.target
            )
            select_mcp_instance(config, session.instance)
        if prepared_stage is not None:
            select_session_stage(
                config, session, args, online=online, prepared_entry=prepared_stage
            )
        return

    if command == "attach":
        if not args.dry_run:
            wait_for_session_mcp(
                config, session.instance, args.timeout, session.card_path, session.profile, session.target
            )
            select_mcp_instance(config, session.instance)
        attach_command, details = session_env_command(config, session, "attach")
        pane_details = respawn_tmux_session_pane(
            args.pane,
            session.card_path,
            attach_command,
            dry_run=args.dry_run,
            force=args.force_pane,
            stem_instance=session.instance,
        )
        print_key_values(pane_details)
        if args.dry_run:
            print_env_plan(details)
        if prepared_stage is not None and not args.dry_run:
            select_session_stage(
                config, session, args, online=online, prepared_entry=prepared_stage
            )
        return
    raise SystemExit(f"Unsupported session command: {command}")


def workspace_info_status(config: WorkspaceConfig) -> str:
    info_root = workspace_info_root(config)
    if (info_root / ".git").is_dir():
        return "git"
    if config.workspace_info_path.is_file():
        return "file"
    return "missing"


def sync_workspace_info(config: WorkspaceConfig, pull_existing: bool = True) -> str:
    info_root = workspace_info_root(config)
    info_root.parent.mkdir(parents=True, exist_ok=True)
    if (info_root / ".git").is_dir():
        if pull_existing:
            subprocess.run(["git", "-C", str(info_root), "pull", "--ff-only"], check=True)
            config.workspace_info = json.loads(config.workspace_info_path.read_text(encoding="utf-8"))
            return "pulled"
        if config.workspace_info_path.is_file():
            config.workspace_info = json.loads(config.workspace_info_path.read_text(encoding="utf-8"))
            return "existing"
        raise SystemExit(f"Workspace info repo exists but manifest is missing: {config.workspace_info_path}")
    if info_root.exists():
        raise SystemExit(f"Workspace info path exists but is not a git repo: {info_root}")

    print(f"workspace-info missing; cloning {config.workspace_info_url}", file=sys.stderr)
    subprocess.run(["git", "clone", config.workspace_info_url, str(info_root)], check=True)
    if not config.workspace_info_path.is_file():
        raise SystemExit(f"Workspace info clone does not contain manifest: {config.workspace_info_path}")
    config.workspace_info = json.loads(config.workspace_info_path.read_text(encoding="utf-8"))
    return "cloned"


def ensure_workspace_info(config: WorkspaceConfig) -> None:
    if config.workspace_info:
        return
    if config.workspace_info_path.is_file():
        config.workspace_info = json.loads(config.workspace_info_path.read_text(encoding="utf-8"))
        return
    sync_workspace_info(config, pull_existing=False)


def run_workspace(config: WorkspaceConfig, args: argparse.Namespace) -> None:
    if args.workspace_command == "show":
        print(f"workspace_info_path\t{config.workspace_info_path}")
        print(f"workspace_info_url\t{config.workspace_info_url}")
        print(f"status\t{workspace_info_status(config)}")
        if config.workspace_info:
            print(f"workspace\t{config.workspace_info.get('workspace', '')}")
            print(f"series\t{len(config.workspace_info.get('series', []))}")
        return

    if args.workspace_command == "sync":
        info_root = workspace_info_root(config)
        if args.dry_run:
            print(f"workspace_info_path\t{config.workspace_info_path}")
            print(f"workspace_info_url\t{config.workspace_info_url}")
            print(f"status\t{workspace_info_status(config)}")
            print(f"command\t{'pull' if (info_root / '.git').is_dir() else 'clone'}")
            return
        sync_workspace_info(config, pull_existing=True)
        print(f"workspace_info_path\t{config.workspace_info_path}")
        print(f"status\tready")
        return

    if args.workspace_command == "audit":
        ensure_workspace_info(config)
        checks = audit_workspace_catalog(config)
        print("check\tstatus\tdetail")
        for name, status, detail in checks:
            print(f"{name}\t{status}\t{detail}")
        if any(status == "FAIL" for _, status, _ in checks):
            raise SystemExit(1)
        return

    if args.workspace_command == "migrate":
        migrated = 0
        already_ready = 0
        candidates = sorted(path for path in config.series_root.glob("*/*") if path.is_dir())
        for card_path in candidates:
            legacy_state = card_path / ".rv"
            stem_state = card_path / ".stem"
            if stem_state.exists() or stem_state.is_symlink():
                already_ready += 1
                continue
            if not legacy_state.is_dir():
                continue
            print(f"state\t{stem_state} -> .rv")
            migrated += 1
            if not args.dry_run:
                stem_state.symlink_to(".rv", target_is_directory=True)
        print(f"migrated\t{migrated}")
        print(f"already_ready\t{already_ready}")
        print(f"mode\t{'dry-run' if args.dry_run else 'applied'}")
        return

    if args.workspace_command == "doctor":
        requested_runtime = os.environ.get("STEM_RUNTIME", "").strip()
        runtime = requested_runtime or ("podman" if shutil.which("podman") else "")
        runtime_ok = runtime == "podman" and shutil.which("podman") is not None
        if requested_runtime and requested_runtime != "podman":
            runtime_detail = f"{requested_runtime} is not supported for the rootless student workflow"
        else:
            runtime_detail = runtime or "missing podman"
        checks: list[tuple[str, bool, str]] = [
            ("workspace", config.workspace_root.is_dir(), str(config.workspace_root)),
            ("series", config.series_root.is_dir(), str(config.series_root)),
            ("env-tool", env_tool_command_path(config).is_file(), str(env_tool_command_path(config))),
            ("runtime", runtime_ok, runtime_detail),
        ]
        socket_probe = config.socket_root / ("t" * 12) / ("i" * 12) / ("c" * 12) / "n.sock"
        checks.append(("socket-path", len(os.fsencode(socket_probe)) <= 100, f"{len(os.fsencode(socket_probe))} bytes"))
        if runtime_ok:
            probe = subprocess.run(
                ["podman", "info", "--format", "{{.Host.Security.Rootless}}"],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                check=False,
            )
            checks.append(("rootless", probe.returncode == 0 and probe.stdout.strip() == "true", probe.stdout.strip() or probe.stderr.strip()))
        else:
            checks.append(("rootless", False, "rootless Podman is required"))
        print("check\tstatus\tdetail")
        for name, ok, detail in checks:
            print(f"{name}\t{'ok' if ok else 'FAIL'}\t{detail}")
        if not all(ok for _, ok, _ in checks):
            raise SystemExit(1)
        return

    raise SystemExit(f"Unsupported workspace command: {args.workspace_command}")


def print_config(config: WorkspaceConfig) -> None:
    print(f"config_path\t{config.config_path}")
    print(f"original_root\t{config.original_root}")
    print(f"original_series_root\t{config.original_series_root}")
    print(f"workspace_root\t{config.workspace_root}")
    print(f"workspace_info_path\t{config.workspace_info_path}")
    print(f"workspace_info_url\t{config.workspace_info_url}")
    print(f"workspace_info_status\t{workspace_info_status(config)}")
    print(f"series_root\t{config.series_root}")
    print(f"socket_root\t{config.socket_root}")
    print(f"token_path\t{config.token_path}")
    print(f"tools_root\t{config.tools_root}")
    print(f"env_tool_url\t{config.env_tool_url}")
    print(f"git_base_url\t{config.git.base_url}")
    print(f"git_source_org\t{config.git.source_org}")
    print(f"git_answer_org\t{config.git.answer_org}")
    print(f"git_source_remote\t{config.git.source_remote}")
    print(f"git_answer_remote\t{config.git.answer_remote}")
    print(f"git_origin_remote\t{config.git.origin_remote}")
    print(f"git_fallback_branch\t{config.git.fallback_branch}")
    print("tools_root_candidates")
    for path in config.tools_root_candidates:
        print(f"  {path}")


def add_repo_option(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        "--repo",
        help="Path inside a target git repo. Default: current working directory.",
    )


def add_store_selection_options(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--server", help="Server endpoint from tokens.json.")
    parser.add_argument("--user", help="API user recorded on the token.")
    parser.add_argument("--token-name", help="Remote id inside tokens.json, for example 'r1'. Default: --remote.")


def print_table(headers: list[str], rows: list[list[str]]) -> None:
    widths = [len(header) for header in headers]
    for row in rows:
        for index, value in enumerate(row):
            widths[index] = max(widths[index], len(value))

    print("  ".join(header.ljust(widths[index]) for index, header in enumerate(headers)))
    print("  ".join("-" * width for width in widths))
    for row in rows:
        print("  ".join(value.ljust(widths[index]) for index, value in enumerate(row)))


def print_config_summary(config_path: Path) -> None:
    try:
        config = load_config(config_path)
    except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
        print("Config:")
        print_table(
            ["field", "value"],
            [
                ["config", str(config_path)],
                ["status", f"unavailable: {error}"],
            ],
        )
        return

    print("Config:")
    print_table(
        ["field", "value"],
        [
            ["config", str(config.config_path)],
            ["workspace", str(config.workspace_root)],
            ["series", str(config.series_root)],
            ["tokens", str(config.token_path)],
        ],
    )


def print_main_overview(config_path: Path) -> None:
    print("STEMCTL - launcher kart pracy STEM")
    print()
    print("Usage:")
    print("  ./stemctl <command> [options]")
    print("  ./stemctl tokens <command> [options]")
    print("  ./stemctl <command> --help")
    print()
    print("Commands:")
    print_table(
        ["command", "arguments", "purpose"],
        [
            ["show-config", "", "show resolved paths and Git defaults"],
            ["workspace", "show|sync|migrate|doctor", "manage and diagnose workspace"],
            ["series", "list|use|cards|fetch|tasks", "work with series, cards and tasks"],
            ["card", "use [series] card", "set the default card, optionally with the default series"],
            ["tasks", "list|show|switch", "shortcut for tasks in the default card"],
            ["env", "list|sync|ensure|build", "build three container profiles from Dockerfile"],
            ["build/test/run/debug", "PROFILE [selector]", "execute a card action"],
            ["deploy", "rp2350 [selector] --target TARGET", "flash a physical RP2350"],
            ["start/status/attach/stop/rm", "PROFILE [selector]", "manage a persistent rootless container"],
            ["probe", "list", "discover RP2350 probes and BOOTSEL targets"],
            ["mcp", "tmux|nvim INSTANCE", "serve one verified container session over stdio"],
            ["session", "choices|status|start|reset|refresh|attach|detach|stage", "select and control a lesson session"],
            ["list-series", "", "compat alias for series list"],
            ["list-cards", "[series]", "compat alias for series cards list"],
            ["tmux-container", "[series] [card]", "compat: run canonical rootless stem shell in pane 0"],
            ["submission", "[series] [card] --class K [--task T]", "prepare answer repo and student branch"],
            ["tokens", "list|compare|cmp|sync|update|remove|rm|add|read|stats|write", "manage tokens.json and Git remote credentials"],
        ],
    )
    print()
    print("Typical flow:")
    print_table(
        ["step", "command", "result"],
        [
            ["1", "./stemctl workspace sync", "clone or update workspace-info"],
            ["2", "./stemctl series cards fetch inf bss", "fetch the card on the host"],
            ["3", "./stemctl env ensure native-amd64", "build or reuse the local native profile"],
            ["4", "./stemctl test native-amd64 1", "run native golden/sanitizer tests"],
            ["5", "./stemctl test hazard3-sim 1", "run the same vectors on Hazard3"],
            ["6", "./stemctl debug hazard3-sim 1", "open common tmux/nvim/GDB UI"],
            ["7", "./stemctl deploy rp2350 1 --target rp2350-rv --device /dev/bus/usb/001/006", "flash during a practical class"],
        ],
    )
    print()
    print_config_summary(config_path)
    print()
    print("Docs:")
    print("  doc/rvctl.md")
    print("  doc/tokens.md")
    print("  doc/tokens.schema.json")
    print("  doc/containers.md")
    print("  doc/workspace.md")
    print("  doc/session.md")


def print_tokens_overview() -> None:
    print("RVCTL tokens - token store and Git remote credentials")
    print()
    print("Usage:")
    print("  ./rvctl tokens <command> [options]")
    print("  ./rvctl tokens <command> --help")
    print()
    print("Commands:")
    print_table(
        ["command", "common options", "direction", "purpose"],
        [
            ["list", "store|remote|both", "read-only", "list one source without comparing it"],
            ["compare/cmp", "[--repo PATH]", "read-only", "compare remote/store with marker and auth masks"],
            ["add", "REMOTE_ID", "tokens.json", "add an empty token skeleton for manual editing"],
            ["read", "[--server ENDPOINT]", "tokens.json", "show servers, remote ids and tokens"],
            ["stats", "[--repo PATH]", "repo + tokens.json", "compare remote URLs with local token store"],
            ["sync remote", "REMOTE_ID [--repo PATH]", "repo -> tokens.json", "copy one remote URL into the store"],
            ["sync store", "REMOTE_ID [--repo PATH]", "tokens.json -> repo", "write one store record into a remote URL"],
            ["remove/rm", "store|remote|both REMOTE_ID", "selected", "remove token store record or Git remote"],
            ["write", "--remote R [--replace]", "tokens.json -> repo", "write selected token into a remote URL"],
            ["update", "REMOTE_ID", "API -> tokens.json", "refresh valid/scope/org/repo metadata"],
        ],
    )
    print()
    print("Examples:")
    print_table(
        ["case", "command"],
        [
            ["list token store", "./rvctl tokens list store"],
            ["list launcher remotes", "./rvctl tokens list remote"],
            ["compare launcher state", "./rvctl tokens cmp"],
            ["read remote r1 into store", "./rvctl tokens sync remote r1"],
            ["refresh r1 metadata", "./rvctl tokens update r1"],
            ["write store r1 to remote", "./rvctl tokens sync store r1 --repo PATH"],
            ["remove r1 from store", "./rvctl tokens rm store r1"],
            ["remove Git remote r1", "./rvctl tokens rm remote r1"],
            ["compare selected card", "./rvctl tokens cmp --repo ~/dev/workspace/rv/series/inf/lab-rv32i-strlen-bss-data-stack"],
            ["stats selected card", "./rvctl tokens stats --repo ~/dev/workspace/rv/series/inf/lab-rv32i-strlen-bss-data-stack"],
            ["write auth to r1", "./rvctl tokens write --repo PATH --remote r1 --server URL"],
        ],
    )


def extract_config_path(argv: list[str]) -> tuple[Path, list[str]]:
    config_path = Path(__file__).with_name("workspace.json")
    remaining: list[str] = []
    index = 0
    while index < len(argv):
        arg = argv[index]
        if arg == "--config":
            if index + 1 >= len(argv):
                remaining.append(arg)
                index += 1
                continue
            config_path = Path(argv[index + 1])
            index += 2
            continue
        if arg.startswith("--config="):
            config_path = Path(arg.split("=", 1)[1])
            index += 1
            continue
        remaining.append(arg)
        index += 1

    return config_path, remaining


def positive_finite_float(raw_value: str) -> float:
    try:
        value = float(raw_value)
    except ValueError as error:
        raise argparse.ArgumentTypeError("must be a number") from error
    if not (value > 0 and value < float("inf")):
        raise argparse.ArgumentTypeError("must be a finite number greater than zero")
    return value


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog=os.environ.get("STEM_CLI_NAME", Path(sys.argv[0]).name),
        description="STEM card launcher for native AMD64, Hazard3 simulation and RP2350.",
    )
    parser.add_argument(
        "--config",
        default=Path(__file__).with_name("workspace.json"),
        type=Path,
        help="Path to workspace.json",
    )

    subparsers = parser.add_subparsers(dest="command", required=True)

    subparsers.add_parser("show-config", help="Print resolved workspace paths.")
    subparsers.add_parser("list-series", help="List series directories from series_root.")

    cards_parser = subparsers.add_parser("list-cards", help="List cards in a series.")
    cards_parser.add_argument("series", nargs="?", help="Series id, for example 'inf'.")

    workspace_parser = subparsers.add_parser("workspace", help="Manage workspace metadata manifest.")
    workspace_subparsers = workspace_parser.add_subparsers(dest="workspace_command", required=True)
    workspace_subparsers.add_parser("show", help="Show workspace-info configuration.")
    workspace_subparsers.add_parser(
        "audit", help="Validate series ownership, organization registry and card identities."
    )
    workspace_sync_parser = workspace_subparsers.add_parser("sync", help="Clone or update workspace-info into meta/.")
    workspace_sync_parser.add_argument("--dry-run", action="store_true", help="Print plan without cloning or pulling.")
    workspace_migrate_parser = workspace_subparsers.add_parser(
        "migrate", help="Expose legacy .rv state as .stem without deleting data."
    )
    workspace_migrate_parser.add_argument("--dry-run", action="store_true", help="Only list state directories.")
    workspace_subparsers.add_parser("doctor", help="Check workspace, rootless runtime and MCP socket budget.")

    env_parser = subparsers.add_parser("env", help="Manage the three canonical container environments.")
    env_subparsers = env_parser.add_subparsers(dest="env_command", required=True)
    env_subparsers.add_parser("list", help="List supported environment profiles.")
    env_sync_parser = env_subparsers.add_parser("sync", help="Clone or update rv32i-hazard3-student-env into tools/.")
    env_sync_parser.add_argument("--dry-run", action="store_true", help="Print plan without cloning or pulling.")
    env_sources_parser = env_subparsers.add_parser(
        "sources", help="Prepare or verify pinned Hazard3 sources in one host card workspace."
    )
    env_sources_parser.add_argument("series", nargs="?", help="Series id; defaults to workspace.json.")
    env_sources_parser.add_argument("card", nargs="?", help="Card id; defaults to workspace.json.")
    env_sources_parser.add_argument("--check", action="store_true", help="Verify without changing the workspace.")
    env_sources_parser.add_argument("--dry-run", action="store_true", help="Print the preparation command only.")
    env_build_parser = env_subparsers.add_parser("build", help="Build one container profile.")
    env_build_parser.add_argument("profile", choices=ENV_PROFILE_CHOICES, help="Environment profile.")
    env_build_parser.add_argument("--dry-run", action="store_true", help="Print command without building.")
    env_ensure_parser = env_subparsers.add_parser("ensure", help="Reuse or locally build a profile from Dockerfile.")
    env_ensure_parser.add_argument("profile", choices=ENV_PROFILE_CHOICES, help="Environment profile.")
    env_ensure_parser.add_argument("--dry-run", action="store_true", help="Print command without building.")

    for action, help_text in (
        ("build", "Build the selected task inside a profile."),
        ("test", "Run the selected task tests inside a profile."),
        ("run", "Run the selected task without implicit deployment."),
        ("debug", "Open the selected task in the common tmux/nvim debugger UI."),
        ("deploy", "Deploy an RP2350 artifact using the explicitly selected probe."),
        ("shell", "Open a shell in the persistent profile container."),
        ("start", "Create or start the persistent profile container."),
        ("status", "Show status of the selected profile container."),
        ("attach", "Attach to the selected profile container UI."),
        ("stop", "Stop the selected profile container."),
        ("rm", "Remove the selected profile container; sources remain on the host."),
    ):
        action_parser = subparsers.add_parser(action, help=help_text)
        action_parser.add_argument("profile", choices=ENV_PROFILE_CHOICES, help="Environment profile.")
        action_parser.add_argument("selector", nargs="*", help="Optional selector: [task], [card task] or [series card task].")
        action_parser.add_argument("--task", help="Task override. Defaults to selector or defaults.task.")
        action_parser.add_argument("--target", help="Target override, for example rp2350-rv or rp2350-arm.")
        action_parser.add_argument("--instance", help="Stable logical container instance name.")
        action_parser.add_argument("--pane", help="Send command to a host tmux pane (for example 0 or %%3).")
        action_parser.add_argument("--dry-run", action="store_true", help="Print command without changing state.")
        if action in {"debug", "shell", "attach"}:
            action_parser.add_argument("--editor", choices=["nvim", "vim"], help="Editor inside the container.")
        if action in {"debug", "deploy"}:
            action_parser.add_argument(
                "--device",
                help="Exact host device path reported by 'stemctl probe list'.",
            )
        if action == "deploy":
            action_parser.add_argument(
                "--backend",
                choices=["probe", "bootsel"],
                default="probe",
                help="Deployment transport (default: probe).",
            )
        if action == "run":
            action_parser.add_argument(
                "--allow-existing-firmware",
                action="store_true",
                help="For RP2350 only: run firmware whose deployment cannot be verified.",
            )

    probe_parser = subparsers.add_parser("probe", help="Discover RP2350 debug probes and BOOTSEL devices.")
    probe_subparsers = probe_parser.add_subparsers(dest="probe_command", required=True)
    probe_list_parser = probe_subparsers.add_parser("list", help="List usable probe/target devices without sudo.")
    probe_list_parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON.")

    mcp_parser = subparsers.add_parser("mcp", help="Select an MCP container or run its bundled MCP server.")
    mcp_subparsers = mcp_parser.add_subparsers(dest="mcp_kind", required=True)
    for kind in ("tmux", "nvim"):
        kind_parser = mcp_subparsers.add_parser(kind, help=f"Serve {kind} tools over MCP stdio.")
        kind_parser.add_argument("instance", help="Stable logical instance name (required to avoid cross-container selection).")
    mcp_select_parser = mcp_subparsers.add_parser(
        "select", help="Atomically point dynamic tmux and Neovim MCP sockets at one container."
    )
    mcp_select_parser.add_argument("selector", help="Logical instance, container name, or at least 12 ID characters.")
    mcp_subparsers.add_parser("status", help="Show and validate the dynamically selected MCP container.")
    mcp_subparsers.add_parser("list", help="List registered container instances and socket readiness.")

    session_parser = subparsers.add_parser(
        "session",
        help="Select a card/task/UML stage and manage its container, MCP and tmux pane.",
    )
    session_subparsers = session_parser.add_subparsers(dest="session_command", required=True)
    session_choices_parser = session_subparsers.add_parser(
        "choices", help="List numbered series, cards, tasks and UML stages with material status."
    )
    session_choices_parser.add_argument("--series", help="Series ID or 1-based number.")
    session_choices_parser.add_argument("--card", help="Card ID or 1-based number shown in this table.")
    session_choices_parser.add_argument("--task", help="Limit UML rows to a task ID or number.")
    session_choices_parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON.")

    session_actions = {
        "status": "Show the resolved container, MCP and current browser/UML state.",
        "start": "Start the selected container and optionally connect it to a host tmux pane.",
        "reset": "Remove and recreate the selected container, then optionally connect it.",
        "refresh": "Reuse the selected container and refresh its host-pane connection.",
        "attach": "Attach the running selected container to a host tmux pane.",
        "detach": "Detach the selected container from the host pane without stopping it.",
        "stage": "Select and replay one UML stage (alias: checkpoint).",
    }
    for session_action, help_text in session_actions.items():
        aliases = ["checkpoint"] if session_action == "stage" else []
        action_parser = session_subparsers.add_parser(session_action, aliases=aliases, help=help_text)
        action_parser.add_argument(
            "profile",
            nargs="?",
            choices=ENV_PROFILE_CHOICES,
            default="hazard3-sim",
            help="Container profile (default: hazard3-sim).",
        )
        action_parser.add_argument("--series", help="Series ID or 1-based number.")
        action_parser.add_argument("--card", help="Card ID or 1-based number shown by session choices.")
        action_parser.add_argument("--task", help="Task ID or number, for example 4.")
        action_parser.add_argument("--target", help="Target override, for example hazard3-baremetal.")
        action_parser.add_argument("--instance", help="Stable logical container instance name.")
        action_parser.add_argument(
            "--pane",
            default="current:0",
            help="Host tmux pane; current:0 means pane 0 of this Codex window, none disables pane changes.",
        )
        action_parser.add_argument(
            "--card-url",
            default=os.environ.get("STEM_CARD_URL", "http://127.0.0.1:8080"),
            help="HTTP endpoint of the selected card server.",
        )
        action_parser.add_argument(
            "--stage",
            help="STEP, PHASE/STEP, BLOCK/PHASE/STEP, canonical TASK/... selector or snapshot_ref.",
        )
        action_parser.add_argument("--first", action="store_true", help="Select the first UML step explicitly.")
        action_parser.add_argument("--block", help="UML block ID or 1-based number.")
        action_parser.add_argument("--phase", help="UML phase ID or 1-based number.")
        action_parser.add_argument("--step", help="Displayed global step number, local phase number or step ID.")
        action_parser.add_argument("--snapshot", help="Snapshot ref or 1-based number in the selection.")
        action_parser.add_argument(
            "--focus",
            choices=["task", "block", "phase", "step", "snapshot"],
            default="step",
            help="Browser keyboard-navigation focus level.",
        )
        action_parser.add_argument(
            "--offline",
            action="store_true",
            help="Select the UML stage without following it in the live container.",
        )
        action_parser.add_argument(
            "--control",
            action="store_true",
            help="Enable browser keyboard control of Neovim together with SYNC.",
        )
        action_parser.add_argument(
            "--timeout",
            type=positive_finite_float,
            default=60,
            help="Seconds to wait for MCP/debugger readiness (default: 60).",
        )
        action_parser.add_argument(
            "--force-pane",
            action="store_true",
            help="Allow replacing a pane marked as owned by another Codex (never the invoking pane).",
        )
        action_parser.add_argument("--dry-run", action="store_true", help="Print the plan without changing state.")
        if session_action == "status":
            action_parser.add_argument("--json", action="store_true", help="Emit machine-readable status JSON.")
            action_parser.add_argument(
                "--strict",
                action="store_true",
                help="Exit non-zero unless MCP and the matching card API are both ready.",
            )

    tasks_parser = subparsers.add_parser("tasks", help="Shortcut for tasks in the default card.")
    tasks_subparsers = tasks_parser.add_subparsers(dest="tasks_command", required=True)

    tasks_short_list_parser = tasks_subparsers.add_parser("list", help="List tasks from the default or selected card.")
    tasks_short_list_parser.add_argument("selector", nargs="*", help="Optional card selector: [card] or [series card].")

    tasks_short_show_parser = tasks_subparsers.add_parser("show", help="Show one task from the default or selected card.")
    tasks_short_show_parser.add_argument("selector", nargs="+", help="Task selector: task, card task, or series card task.")

    tasks_short_switch_parser = tasks_subparsers.add_parser("switch", help="Create or switch to the answer branch for a task.")
    tasks_short_switch_parser.add_argument("selector", nargs="+", help="Task selector: task, card task, or series card task.")
    tasks_short_switch_parser.add_argument("--branch", help="Override the branch name.")
    tasks_short_switch_parser.add_argument("--dry-run", action="store_true", help="Print plan without switching branches.")

    card_parser = subparsers.add_parser("card", help="Manage the default card selector.")
    card_subparsers = card_parser.add_subparsers(dest="card_command", required=True)
    card_use_parser = card_subparsers.add_parser("use", help="Set the default card, optionally with the default series.")
    card_use_parser.add_argument("selector", nargs="+", help="Use '<card>' or '<series> <card>', for example 'bss' or 'inf bss'.")

    tmux_parser = subparsers.add_parser(
        "tmux-container",
        help="Compatibility wrapper: create host tmux with the canonical rootless stem shell in pane 0.",
    )
    tmux_parser.add_argument("series", nargs="?", help="Series id or full selector like 'inf/bss'.")
    tmux_parser.add_argument("card", nargs="?", help="Card id or unique fragment, for example 'bss'.")
    tmux_parser.add_argument("--profile", choices=ENV_PROFILE_CHOICES, default="hazard3-sim", help="Container profile.")
    tmux_parser.add_argument("--session", help="tmux session name.")
    tmux_parser.add_argument("--window", help="tmux window name.")
    tmux_parser.add_argument("--instance", help="Stable STEM_INSTANCE passed to the canonical runtime.")
    tmux_parser.add_argument("--attach", action="store_true", help="Attach to the session after creation.")
    tmux_parser.add_argument("--dry-run", action="store_true", help="Print commands instead of running tmux.")

    submission_parser = subparsers.add_parser(
        "submission",
        help="Print or apply source/answer git remotes for a selected card.",
    )
    submission_parser.add_argument("series", nargs="?", help="Series id or full selector like 'inf/bss'.")
    submission_parser.add_argument("card", nargs="?", help="Card id or unique fragment, for example 'bss'.")
    submission_parser.add_argument("--class", dest="class_name", required=True, help="Class id, for example 'c2025-1a-inf'.")
    submission_parser.add_argument("--nick", help="Student nick. Defaults to the user from tokens.json.")
    submission_parser.add_argument("--task", help="Task name or unique fragment. Defaults to defaults.task.")
    submission_parser.add_argument(
        "--branch",
        help="Override the submission branch name. Defaults to <store-user>T<task-number>a.",
    )
    submission_parser.add_argument(
        "--date",
        default=dt.date.today().isoformat(),
        help="Lesson date reported in the submission plan. Default: today in YYYY-MM-DD.",
    )
    submission_parser.add_argument(
        "--source-url",
        help="Override the source repo URL. By default launcher reads r1 or origin from the card repo.",
    )
    submission_parser.add_argument(
        "--apply",
        action="store_true",
        help="Write remotes into the selected card repo instead of only printing the plan.",
    )

    series_parser = subparsers.add_parser("series", help="Work with series, cards and tasks.")
    series_subparsers = series_parser.add_subparsers(dest="series_command", required=True)

    series_subparsers.add_parser("list", help="List available series.")

    series_use_parser = series_subparsers.add_parser("use", help="Set the default series.")
    series_use_parser.add_argument("series", help="Series id, for example 'inf'.")

    series_show_parser = series_subparsers.add_parser("show", help="Show one series.")
    series_show_parser.add_argument("series", help="Series id, for example 'inf'.")

    series_fetch_parser = series_subparsers.add_parser("fetch", help="Fetch all cards in a series.")
    series_fetch_parser.add_argument("series", help="Series id, for example 'inf'.")
    series_fetch_parser.add_argument("--dry-run", action="store_true", help="Print plan without cloning.")

    cards_group_parser = series_subparsers.add_parser("cards", help="Work with cards in a series.")
    cards_subparsers = cards_group_parser.add_subparsers(dest="cards_command", required=True)

    cards_list_parser = cards_subparsers.add_parser("list", help="List cards in a series.")
    cards_list_parser.add_argument("series", help="Series id, for example 'inf'.")

    cards_show_parser = cards_subparsers.add_parser("show", help="Show one card.")
    cards_show_parser.add_argument("series", help="Series id, for example 'inf'.")
    cards_show_parser.add_argument("card", help="Card id or unique fragment, for example 'bss'.")

    cards_fetch_parser = cards_subparsers.add_parser("fetch", help="Fetch one card into workspace.")
    cards_fetch_parser.add_argument("series", help="Series id, for example 'inf'.")
    cards_fetch_parser.add_argument("card", help="Card id or unique fragment, for example 'bss'.")
    cards_fetch_parser.add_argument("--dry-run", action="store_true", help="Print plan without cloning.")

    cards_submission_parser = cards_subparsers.add_parser("submission", help="Print/apply answer remotes for a card.")
    cards_submission_parser.add_argument("series", help="Series id, for example 'inf'.")
    cards_submission_parser.add_argument("card", help="Card id or unique fragment, for example 'bss'.")
    cards_submission_parser.add_argument("--class", dest="class_name", required=True, help="Class id.")
    cards_submission_parser.add_argument("--nick", help="Student nick. Defaults to the user from tokens.json.")
    cards_submission_parser.add_argument("--task", help="Task name or unique fragment. Defaults to defaults.task.")
    cards_submission_parser.add_argument("--branch", help="Override the submission branch name.")
    cards_submission_parser.add_argument("--date", default=dt.date.today().isoformat(), help="Lesson date in YYYY-MM-DD.")
    cards_submission_parser.add_argument("--source-url", help="Override source repo URL.")
    cards_submission_parser.add_argument("--apply", action="store_true", help="Write remotes into the selected card repo.")

    tasks_group_parser = cards_subparsers.add_parser("tasks", help="Work with tasks in a card.")
    tasks_subparsers = tasks_group_parser.add_subparsers(dest="tasks_command", required=True)

    tasks_list_parser = tasks_subparsers.add_parser("list", help="List tasks in a card.")
    tasks_list_parser.add_argument("series", help="Series id, for example 'inf'.")
    tasks_list_parser.add_argument("card", help="Card id or unique fragment, for example 'bss'.")

    tasks_show_parser = tasks_subparsers.add_parser("show", help="Show one task.")
    tasks_show_parser.add_argument("series", help="Series id, for example 'inf'.")
    tasks_show_parser.add_argument("card", help="Card id or unique fragment, for example 'bss'.")
    tasks_show_parser.add_argument("task", help="Task name or unique fragment.")

    tasks_switch_parser = tasks_subparsers.add_parser("switch", help="Create or switch to the answer branch for a task.")
    tasks_switch_parser.add_argument("series", help="Series id, for example 'inf'.")
    tasks_switch_parser.add_argument("card", help="Card id or unique fragment, for example 'bss'.")
    tasks_switch_parser.add_argument("task", help="Task name or unique fragment, for example 'task4' or '4'.")
    tasks_switch_parser.add_argument("--branch", help="Override the branch name.")
    tasks_switch_parser.add_argument("--dry-run", action="store_true", help="Print plan without switching branches.")

    tokens_parser = subparsers.add_parser("tokens", help="Manage local token store and repo remotes.")
    tokens_subparsers = tokens_parser.add_subparsers(dest="tokens_command", required=True)

    tokens_compare_parser = tokens_subparsers.add_parser(
        "compare",
        aliases=["cmp"],
        help="Compare repo remotes with tokens.json and print paired token rows.",
    )
    tokens_compare_parser.add_argument(
        "--repo",
        help="Path inside a target git repo. Default: repo containing rvctl.",
    )

    tokens_list_parser = tokens_subparsers.add_parser(
        "list",
        help="List token store records, Git remotes, or both without comparing them.",
    )
    tokens_list_parser.add_argument("target", choices=["store", "remote", "both"], help="Source to list.")
    tokens_list_parser.add_argument(
        "--repo",
        help="Path inside a target git repo for target remote/both. Default: repo containing rvctl.",
    )
    tokens_list_parser.add_argument("--server", help="Filter output to one server endpoint.")

    tokens_read_parser = tokens_subparsers.add_parser(
        "read",
        help="Read tokens.json and print stored servers, token ids and remotes.",
    )
    tokens_read_parser.add_argument("--server", help="Filter output to one server endpoint.")
    tokens_read_parser.add_argument("--show-secrets", action="store_true", help="Print full token values.")

    tokens_add_parser = tokens_subparsers.add_parser(
        "add",
        help="Add an empty token skeleton to tokens.json.",
    )
    tokens_add_parser.add_argument("token_id", metavar="REMOTE_ID", help="Remote id, for example 'r1' or 'r1a'.")
    tokens_add_parser.add_argument("--server", help="Server endpoint. Default: git.base_url from workspace.json.")
    tokens_add_parser.add_argument("--value", default="", help="Optional token value. Default: empty for manual editing.")
    tokens_add_parser.add_argument("--user", help="Optional API user label.")
    tokens_add_parser.add_argument("--remote", help="Compatibility alias. If used, it must match REMOTE_ID.")
    tokens_add_parser.add_argument("--org", help="Optional remote organization.")
    tokens_add_parser.add_argument("--repo", dest="repo_name", help="Optional remote repository.")
    tokens_add_parser.add_argument("--dry-run", action="store_true", help="Print the planned token without writing it.")

    tokens_stats_parser = tokens_subparsers.add_parser(
        "stats",
        help="Print per-server token statistics from tokens.json.",
    )
    tokens_stats_parser.add_argument(
        "--repo",
        help="Path inside a target git repo. Default: repo containing rvctl.",
    )
    tokens_stats_parser.add_argument("--server", help="Filter output to one server endpoint.")

    tokens_sync_parser = tokens_subparsers.add_parser(
        "sync",
        help="Synchronize one remote id between git remote URL and tokens.json.",
    )
    tokens_sync_parser.add_argument("direction", choices=["remote", "store"], help="'remote' reads git remote into store; 'store' writes store into git remote.")
    tokens_sync_parser.add_argument("remote_id", help="Remote id, for example 'r1' or 'r1a'.")
    tokens_sync_parser.add_argument(
        "--repo",
        help="Path inside a target git repo. Default: repo containing rvctl.",
    )
    tokens_sync_parser.add_argument("--url", help="Override target remote URL when writing store -> remote.")
    add_store_selection_options(tokens_sync_parser)
    tokens_sync_parser.add_argument("--replace", action="store_true", help="Replace different credentials when writing store -> remote.")
    tokens_sync_parser.add_argument("--dry-run", action="store_true", help="Print the planned sync without writing it.")

    tokens_remove_parser = tokens_subparsers.add_parser(
        "remove",
        aliases=["rm"],
        help="Remove a token store record, a Git remote, or both.",
    )
    tokens_remove_parser.add_argument("target", choices=["store", "remote", "both"], help="Where to remove REMOTE_ID from.")
    tokens_remove_parser.add_argument("remote_id", help="Remote id, for example 'r1' or 'r1a'.")
    tokens_remove_parser.add_argument(
        "--repo",
        help="Path inside a target git repo for target remote/both. Default: repo containing rvctl.",
    )
    tokens_remove_parser.add_argument("--server", help="Server endpoint used to disambiguate store records.")
    tokens_remove_parser.add_argument("--dry-run", action="store_true", help="Print the planned removal without writing it.")

    tokens_write_parser = tokens_subparsers.add_parser(
        "write",
        help="Write credentials from tokens.json into a repo remote URL.",
    )
    add_repo_option(tokens_write_parser)
    tokens_write_parser.add_argument("--remote", help="Remote name to update or create.")
    tokens_write_parser.add_argument("--url", help="Override target remote URL when creating or updating the remote.")
    add_store_selection_options(tokens_write_parser)
    tokens_write_parser.add_argument("--replace", action="store_true", help="Replace different credentials in the remote URL.")
    tokens_write_parser.add_argument("--dry-run", action="store_true", help="Print the planned update without writing it.")

    tokens_update_parser = tokens_subparsers.add_parser(
        "update",
        help="Refresh API metadata for one remote id, or use a legacy --from direction.",
    )
    tokens_update_parser.add_argument("remote_id", nargs="?", help="Remote id to refresh from API, for example 'r1'.")
    tokens_update_parser.add_argument(
        "--from",
        dest="from_source",
        choices=["remotes", "store"],
        help="Legacy synchronization direction source: 'remotes' or 'store'.",
    )
    add_repo_option(tokens_update_parser)
    tokens_update_parser.add_argument("--remote", help="Remote name used when --from store.")
    tokens_update_parser.add_argument("--url", help="Override target remote URL when --from store.")
    add_store_selection_options(tokens_update_parser)
    tokens_update_parser.add_argument("--replace", action="store_true", help="Replace different credentials when writing to remotes.")
    tokens_update_parser.add_argument("--dry-run", action="store_true", help="Print the planned update without writing it.")

    return parser


def main() -> int:
    config_path, command_args = extract_config_path(sys.argv[1:])
    if not command_args or command_args == ["help"]:
        print_main_overview(config_path)
        return 0
    if command_args in (["tokens"], ["help", "tokens"], ["tokens", "help"]):
        print_tokens_overview()
        return 0

    parser = build_parser()
    args = parser.parse_args()
    config = load_config(args.config)
    profile_actions = {"build", "test", "run", "debug", "deploy", "shell", "start", "status", "attach", "stop", "rm"}
    if args.command in {"list-series", "list-cards", "series", "card", "tasks", "session", "tmux-container", "submission"} | profile_actions:
        ensure_workspace_info(config)

    if args.command == "show-config":
        print_config(config)
        return 0
    if args.command == "list-series":
        print_series(config)
        return 0
    if args.command == "list-cards":
        print_cards(config, args.series)
        return 0
    if args.command == "workspace":
        run_workspace(config, args)
        return 0
    if args.command == "env":
        run_env(config, args)
        return 0
    if args.command == "probe":
        run_probe(config, args)
        return 0
    if args.command == "mcp":
        run_mcp(config, args)
        return 0
    if args.command == "session":
        run_session(config, args)
        return 0
    if args.command in profile_actions:
        run_profile_action(config, args, args.command)
        return 0
    if args.command == "tasks":
        run_tasks(config, args)
        return 0
    if args.command == "card":
        run_card(config, args)
        return 0
    if args.command == "tmux-container":
        run_tmux_container(config, args)
        return 0
    if args.command == "submission":
        print_submission_plan(config, args)
        return 0
    if args.command == "series":
        run_series(config, args)
        return 0
    if args.command == "tokens":
        if args.tokens_command in {"compare", "cmp"}:
            run_tokens_compare(config, args)
            return 0
        if args.tokens_command == "list":
            run_tokens_list(config, args)
            return 0
        if args.tokens_command == "add":
            run_tokens_add(config, args)
            return 0
        if args.tokens_command == "read":
            run_tokens_read(config, args)
            return 0
        if args.tokens_command == "stats":
            run_tokens_stats(config, args)
            return 0
        if args.tokens_command == "sync":
            run_tokens_sync(config, args)
            return 0
        if args.tokens_command in {"remove", "rm"}:
            run_tokens_remove(config, args)
            return 0
        if args.tokens_command == "write":
            run_tokens_write(config, args)
            return 0
        if args.tokens_command == "update":
            run_tokens_update(config, args)
            return 0
        parser.error(f"Unsupported tokens command: {args.tokens_command}")
        return 2

    parser.error(f"Unsupported command: {args.command}")
    return 2


if __name__ == "__main__":
    sys.exit(main())