diff --git a/README.md b/README.md index 7a98270..e35464d 100644 --- a/README.md +++ b/README.md @@ -59,6 +59,21 @@ pozostają pod `~/dev/workspace/stem`. Kontener dostaje wybraną kartę jako Pierwsze wywołanie danego profilu może potrwać, ponieważ buduje go ze źródłowego Dockerfile. Kolejne korzystają z lokalnych warstw cache. +Pełną sesję lekcji można wybrać i kontrolować numerami z CLI: + +```bash +./stemctl session choices --series inf --card 7 --task 4 +./stemctl session reset hazard3-sim --series inf --card 7 --task 4 --step 12 +./stemctl session refresh hazard3-sim --series inf --card 7 --task 4 +./stemctl session attach hazard3-sim --series inf --card 7 --task 4 +./stemctl session detach hazard3-sim --series inf --card 7 --task 4 +``` + +`current:0` jest domyślnym pane: pane `0` okna tmuxa, w którym działa +wywołujący Codex. `session stage`/`session checkpoint` ustawia dowolny etap UML +online w Termdebug albo tylko na stronie z `--offline`. Szczegóły i wszystkie +selektory opisuje [dokumentacja sesji](doc/session.md). + Wymuszenie samego przygotowania środowiska: ```bash @@ -203,6 +218,7 @@ Nowe materiały powinny używać nazw kanonicznych. - [plan architektury](doc/architecture-plan.md) - [kontenery i interfejs](doc/containers.md) - [wytyczne dokumentowania debugowania](doc/debug-documentation-guidelines.md) +- [sesja karty, UML, MCP i tmux z CLI](doc/session.md) - [migracja nazw i workspace](doc/migration-stem-launcher.md) - [serie i karty](doc/series.md) - [tokeny Gitea](doc/tokens.md) diff --git a/doc/session.md b/doc/session.md new file mode 100644 index 0000000..49ebae4 --- /dev/null +++ b/doc/session.md @@ -0,0 +1,113 @@ +# Sesja karty z wiersza poleceń + +`stemctl session` łączy pięć selektorów w jeden kontrakt: + +```text +seria → karta → Task → blok/faza/krok/snapshot UML → instancja kontenera +``` + +Każdy selektor przyjmuje stabilne ID. Seria, karta, Task, blok, faza, krok i +snapshot przyjmują także numery pokazywane przez `choices`. Numer karty jest +jednobazową pozycją w manifeście serii, więc nie zależy od tego, czy repo karty +zostało już pobrane. W przypadku kroków najpierw sprawdzany jest globalny numer +strzałki z diagramu, a przy jawnie wybranej fazie także lokalny numer kroku. + +## Lista wyboru + +```bash +./stemctl session choices --series inf --card 7 --task 4 +./stemctl session choices --series inf --card pointers --json +``` + +Tabele zawierają kolumnę `status`: + +- `opracowane` — materiał UML ma komplet recept replay/checkpoint; +- `robocze` — istnieje źródło albo częściowe metadane; +- `brak` — karta lub materiał nie znajduje się jeszcze w workspace. + +## Sterowanie środowiskiem + +```bash +# Utwórz debugger Task04 w pane 0 okna, z którego uruchomiono Codexa. +./stemctl session start hazard3-sim --series inf --card 7 --task 4 + +# Usuń i odtwórz kontener, po czym zatrzymaj maszynę na strzałce 12. +./stemctl session reset hazard3-sim \ + --series inf --card 7 --task 4 --step 12 + +# Zachowaj kontener i ponownie podepnij istniejący debugger. +./stemctl session refresh hazard3-sim --series inf --card 7 --task 4 + +# Podepnij/odepnij tylko klienta w zewnętrznym tmuxie. Serwer debuggera żyje dalej. +./stemctl session attach hazard3-sim --series inf --card 7 --task 4 +./stemctl session detach hazard3-sim --series inf --card 7 --task 4 +``` + +Domyślny cel `--pane current:0` oznacza pane `0` tego samego okna tmuxa, w +którym działa wywołujący Codex. Można podać jawne `%ID`, `sesja:okno.pane` albo +`--pane none`. Launcher odmawia zastąpienia pane, z którego sam został +uruchomiony, aby nie zakończyć Codexa. Pane Codexa jest dodatkowo oznaczone +PID-em pane w opcjach tmuxa; jawne `%ID` podane z innego terminala również +zostanie odrzucone. Świadome obejście tej drugiej ochrony wymaga +`--force-pane`; nie da się nim zastąpić pane wykonującego bieżącą komendę. + +`detach` odłącza wyłącznie zewnętrznego klienta tmuxa. Nie zatrzymuje +kontenera, symulatora, GDB, Neovima ani wewnętrznej sesji tmuxa. + +## Wybór stanu UML + +Równoważne przykłady: + +```bash +./stemctl session stage hazard3-sim --series inf --card 7 --task 4 --step 12 +./stemctl session checkpoint hazard3-sim --series inf --card pointers --task 4 \ + --snapshot task04.alloc5.commit +./stemctl session stage hazard3-sim --series inf --card 7 --task 4 --stage alloc-5/7 +./stemctl session stage hazard3-sim --series inf --card 7 --task 4 \ + --stage task04/allocator-flow/alloc-5/alloc5-commit +./stemctl session stage hazard3-sim --series inf --card 7 --task 4 \ + --phase 2 --step 3 +./stemctl session stage hazard3-sim --series inf --card 7 --task 4 --first +``` + +W ostatnim przykładzie `3` oznacza trzeci krok wewnątrz drugiej fazy, jeżeli +w tej fazie nie istnieje globalna strzałka numer 3. + +Tryb online najpierw sprawdza pełną tożsamość karty (`id`, UUID, wersję i hash +źródłowego JSON-a), receptę checkpointu, profil, target oraz bind mount +`/workspace`. Następnie wybiera sockety MCP i przekazuje kontrolerowi +oczekiwane `instance/container_id/profile/target`. Replay zostaje odrzucony, +jeżeli globalny wybór MCP zmieni się przed aktywacją albo podczas niej. Dopiero +po udanym zatrzymaniu GDB publikowane są pozycja strony i `SYNC ON`. + +`--offline` zmienia wyłącznie stan strony i wyłącza SYNC/sterowanie Neovimem. +`--control` dodatkowo uzbraja sterowanie klawiaturą Neovima; samo SYNC nie robi +tego automatycznie. `stage/checkpoint` wymaga jawnego selektora; pierwszy etap +wybiera się świadomie przez `--first`. + +```bash +./stemctl session stage hazard3-sim --series inf --card 7 --task 4 \ + --step 12 --offline +./stemctl session status hazard3-sim --series inf --card 7 --task 4 +./stemctl session status hazard3-sim --series inf --card 7 --task 4 \ + --json --strict +``` + +Adres serwera karty można zmienić przez `--card-url` albo `STEM_CARD_URL`. +Limit oczekiwania na Neovima/GDB/MCP ustawia `--timeout`. + +## Przyszłe strategie i adnotacje + +Kluczem rozszerzeń pozostaje pełna pozycja +`task/block/phase/step/snapshot`. Pod tym kluczem będzie można później +przechowywać wiele strategii debugowania i adnotacji bez zmiany obecnego CLI. +Planowany kontrakt rozdziela treść od widoczności: + +```text +strategy: id, label, commands, expected_observations +annotation: id, strategy_id, target, geometry, style, text, visible +``` + +`target` może wskazywać diagram, bufor/wiersz Neovima, rejestr, ramkę stosu +albo zakres pamięci. Operacje `show/hide/toggle` mają zmieniać widoczność bez +usuwania adnotacji; `add/remove` będą osobnymi, audytowalnymi operacjami. diff --git a/rvctl.py b/rvctl.py index 99c1451..c408f77 100755 --- a/rvctl.py +++ b/rvctl.py @@ -6,6 +6,7 @@ from __future__ import annotations import argparse import base64 import datetime as dt +import hashlib import json import os import re @@ -14,6 +15,7 @@ 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 @@ -80,6 +82,17 @@ class CardInfo: 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"), @@ -146,6 +159,11 @@ PROFILE_DEFAULT_TARGETS = { "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: @@ -3939,7 +3957,15 @@ def probe_mcp_socket_servers(record: dict, nvim_socket: Path, tmux_socket: Path) def validated_mcp_record(record: dict) -> tuple[dict, Path, Path]: - required = ("runtime", "container_name", "container_id", "instance_key", "socket_dir") + 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', '?')}") @@ -3966,7 +3992,14 @@ def validated_mcp_record(record: dict) -> tuple[dict, Path, Path]: actual_id = str(inspection.get("Id", "")) labels = inspection.get("Config", {}).get("Labels") or {} actual_instance_key = str(labels.get("edu.stem.instance-key", "")) - if actual_id != expected_id or actual_instance_key != expected_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"])) @@ -4117,6 +4150,1240 @@ def run_profile_action(config: WorkspaceConfig, args: argparse.Namespace, action 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)) + 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, +) -> 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) + 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, + ) + 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', '')} != {profile}." + ) + if target is not None and str(record.get("target", "")) != target: + raise SystemExit( + f"MCP target mismatch: {record.get('target', '')} != {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, + ) + 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, + ) + 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, + ) + 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, + ) + 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, + ) + 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(): @@ -4338,6 +5605,7 @@ def print_main_overview(config_path: Path) -> None: ["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"], @@ -4368,6 +5636,7 @@ def print_main_overview(config_path: Path) -> None: print(" doc/tokens.schema.json") print(" doc/containers.md") print(" doc/workspace.md") + print(" doc/session.md") def print_tokens_overview() -> None: @@ -4437,6 +5706,16 @@ def extract_config_path(argv: list[str]) -> tuple[Path, list[str]]: 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), @@ -4546,6 +5825,98 @@ def build_parser() -> argparse.ArgumentParser: 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 debugger UI in a host tmux pane.", + "reset": "Remove and recreate the container, then start its debugger UI.", + "refresh": "Reuse the container and reattach or recreate its debugger UI.", + "attach": "Attach an existing container debugger UI to a host tmux pane.", + "detach": "Detach the host pane without stopping the container debugger.", + "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) @@ -4794,7 +6165,7 @@ def main() -> int: 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", "tmux-container", "submission"} | profile_actions: + 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": @@ -4818,6 +6189,9 @@ def main() -> int: 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 diff --git a/tests/test_stemctl.py b/tests/test_stemctl.py index 9f9f1fa..716d0ac 100644 --- a/tests/test_stemctl.py +++ b/tests/test_stemctl.py @@ -7,7 +7,7 @@ import socket import subprocess import tempfile import unittest -from contextlib import redirect_stdout +from contextlib import redirect_stderr, redirect_stdout from pathlib import Path from types import SimpleNamespace from unittest import mock @@ -33,7 +33,7 @@ class StemctlContractTests(unittest.TestCase): def test_new_command_help_never_crashes(self) -> None: script = Path(__file__).resolve().parents[1] / "rvctl.py" - commands = ["env", "build", "test", "run", "debug", "deploy", "shell", "start", "status", "attach", "stop", "rm", "probe", "mcp"] + commands = ["env", "build", "test", "run", "debug", "deploy", "shell", "start", "status", "attach", "stop", "rm", "probe", "mcp", "session"] for command in commands: with self.subTest(command=command): completed = subprocess.run( @@ -342,6 +342,320 @@ class StemctlContractTests(unittest.TestCase): nvim_server.close() tmux_server.close() + def test_session_parser_exposes_numbered_target_and_stage_controls(self) -> None: + parser = rvctl.build_parser() + args = parser.parse_args( + [ + "session", + "reset", + "hazard3-sim", + "--series", + "1", + "--card", + "7", + "--task", + "4", + "--phase", + "2", + "--step", + "3", + "--pane", + "current:0", + ] + ) + self.assertEqual(args.session_command, "reset") + self.assertEqual(args.card, "7") + self.assertEqual(args.task, "4") + self.assertEqual(args.phase, "2") + self.assertEqual(args.step, "3") + self.assertEqual(args.pane, "current:0") + + def test_session_phase_local_step_falls_back_after_global_number(self) -> None: + entries = [] + for phase_index, numbers in enumerate(((1, 2), (6, 7, 8))): + for step_index, number in enumerate(numbers): + entries.append( + { + "task": {"id": "task04", "label": "Task04", "index": 0}, + "block": {"id": "flow", "label": "Flow", "index": 0}, + "phase": { + "id": f"phase-{phase_index + 1}", + "label": f"Phase {phase_index + 1}", + "index": phase_index, + }, + "step": { + "id": f"step-{number}", + "label": f"Step {number}", + "number": number, + "index": step_index, + "global_index": len(entries), + }, + "snapshot": {"ref": f"task04.step-{number}", "index": step_index}, + } + ) + args = SimpleNamespace( + block=None, + phase="2", + step="3", + snapshot=None, + stage=None, + ) + selected = rvctl.resolve_navigation_entry({"entries": entries}, "task04_address_arithmetic_alloc", args) + self.assertEqual(selected["phase"]["id"], "phase-2") + self.assertEqual(selected["step"]["number"], 8) + + def test_session_material_status_has_three_polish_values(self) -> None: + entry = { + "step": {"id": "one"}, + "snapshot": {"ref": "task04.one"}, + } + self.assertEqual( + rvctl.navigation_entry_status( + entry, + { + "task04.one": { + "stop": {"symbol": "main", "offset": 0}, + "verify": {"expressions": []}, + } + }, + ), + "opracowane", + ) + self.assertEqual(rvctl.navigation_entry_status(entry, {}), "robocze") + self.assertEqual(rvctl.development_status([]), "brak") + + def test_session_card_number_uses_manifest_order(self) -> None: + cards = [ + rvctl.CardInfo("inf", "first", "first", "main", "", "a", "b", "r1", "r1a", "main"), + rvctl.CardInfo("inf", "second", "second", "main", "", "a", "b", "r1", "r1a", "main"), + ] + config = SimpleNamespace(defaults={}) + with mock.patch.object(rvctl, "source_card_infos", return_value=cards): + self.assertEqual(rvctl.resolve_session_card(config, "inf", "2"), "second") + + def test_session_rejects_destructive_online_reset_without_pane_in_preflight(self) -> None: + parser = rvctl.build_parser() + args = parser.parse_args( + [ + "session", + "reset", + "hazard3-sim", + "--series", + "inf", + "--card", + "7", + "--task", + "4", + "--pane", + "none", + "--step", + "12", + ] + ) + with self.assertRaisesRegex(SystemExit, "requires a debugger pane"): + rvctl.validate_session_command_args(args) + + def test_session_rejects_profile_target_pair_before_mutation(self) -> None: + args = SimpleNamespace( + series="inf", + card="7", + task="4", + profile="hazard3-sim", + target="rp2350-rv", + instance=None, + ) + with ( + mock.patch.object(rvctl, "resolve_session_series", return_value="inf"), + mock.patch.object(rvctl, "resolve_session_card", return_value="pointers"), + mock.patch.object(rvctl, "resolve_workspace_card_path", return_value=Path("/tmp/card")), + mock.patch.object(rvctl, "resolve_submission_task", return_value="task04"), + self.assertRaisesRegex(SystemExit, "does not belong to profile hazard3-sim"), + ): + rvctl.resolve_session_target(SimpleNamespace(), args) + + def test_session_validates_pane_before_container_commands(self) -> None: + parser = rvctl.build_parser() + args = parser.parse_args( + ["session", "reset", "hazard3-sim", "--series", "inf", "--card", "7", "--task", "4"] + ) + target = rvctl.SessionTarget( + "inf", "pointers", Path("/tmp/card"), "task04_example", "hazard3-sim", + "hazard3-baremetal", "hazard3-sim-inf-pointers-t4" + ) + with ( + mock.patch.object(rvctl, "resolve_session_target", return_value=target), + mock.patch.object(rvctl, "resolve_tmux_session_pane", side_effect=SystemExit("protected")), + mock.patch.object(subprocess, "run") as run, + self.assertRaisesRegex(SystemExit, "protected"), + ): + rvctl.run_session(SimpleNamespace(), args) + run.assert_not_called() + + def test_session_start_pane_none_applies_offline_stage(self) -> None: + parser = rvctl.build_parser() + args = parser.parse_args( + [ + "session", "start", "hazard3-sim", "--series", "inf", "--card", "7", + "--task", "4", "--pane", "none", "--step", "12", "--offline" + ] + ) + target = rvctl.SessionTarget( + "inf", "pointers", Path("/tmp/card"), "task04_example", "hazard3-sim", + "hazard3-baremetal", "hazard3-sim-inf-pointers-t4" + ) + prepared = { + "task": {"id": "task04"}, + "block": {"id": "flow"}, + "phase": {"id": "alloc-5"}, + "step": {"id": "commit", "number": 12}, + "snapshot": {"ref": "task04.alloc5.commit"}, + } + with ( + mock.patch.object(rvctl, "resolve_session_target", return_value=target), + mock.patch.object(rvctl, "prepare_session_stage", return_value=prepared), + mock.patch.object(rvctl, "session_env_command", return_value=("true", {"command": "true"})), + mock.patch.object(rvctl, "select_session_stage") as select_stage, + mock.patch.object(subprocess, "run"), + redirect_stdout(io.StringIO()), + ): + rvctl.run_session(SimpleNamespace(), args) + select_stage.assert_called_once_with( + mock.ANY, target, args, online=False, prepared_entry=prepared + ) + + def test_session_stage_requires_explicit_selector_and_finite_timeout(self) -> None: + parser = rvctl.build_parser() + args = parser.parse_args( + ["session", "stage", "hazard3-sim", "--series", "inf", "--card", "7", "--task", "4"] + ) + with self.assertRaisesRegex(SystemExit, "requires a UML selector"): + rvctl.validate_session_command_args(args) + with redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + parser.parse_args( + [ + "session", "stage", "hazard3-sim", "--series", "inf", "--card", "7", + "--task", "4", "--first", "--timeout", "nan" + ] + ) + + def test_session_protects_codex_owned_pane_by_pid_marker(self) -> None: + completed = subprocess.CompletedProcess( + [], 0, stdout="1:0.0\t%3\t/tmp\t999\t999\tthread-1\n", stderr="" + ) + with ( + mock.patch.object(rvctl, "mark_invoking_codex_pane"), + mock.patch.object(subprocess, "run", return_value=completed), + mock.patch.dict(os.environ, {"TMUX_PANE": "%33"}), + ): + with self.assertRaisesRegex(SystemExit, "owned by Codex"): + rvctl.resolve_tmux_session_pane("%3") + self.assertEqual(rvctl.resolve_tmux_session_pane("%3", force=True)[1], "%3") + + def test_session_identity_is_fail_closed_on_uuid_or_hash_mismatch(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + card_path = Path(temporary) + (card_path / "json").mkdir() + source = { + "card": { + "id": "card-1", + "uuid": "uuid-1", + "version": "v1", + } + } + (card_path / "json/card_source.json").write_text(json.dumps(source), encoding="utf-8") + target = rvctl.SessionTarget( + "inf", "card", card_path, "task04", "hazard3-sim", + "hazard3-baremetal", "instance" + ) + served = { + "id": "card-1", + "uuid": "different", + "version": "v1", + "source_sha256": "different", + } + with mock.patch.object(rvctl, "card_api_json", return_value=served): + with self.assertRaisesRegex(SystemExit, "uuid, source_sha256"): + rvctl.verify_card_api_identity("http://127.0.0.1:8080", target, 1) + + def test_session_cancelled_checkpoint_is_not_published(self) -> None: + target = rvctl.SessionTarget( + "inf", "pointers", Path("/tmp/card"), "task04", "hazard3-sim", + "hazard3-baremetal", "hazard3-sim-inf-pointers-t4" + ) + args = SimpleNamespace( + card_url="http://127.0.0.1:8080", + timeout=1, + focus="step", + control=False, + ) + entry = { + "task": {"id": "task04"}, + "block": {"id": "allocator-flow"}, + "phase": {"id": "alloc-5"}, + "step": {"id": "alloc5-commit", "number": 12}, + "snapshot": {"ref": "task04.alloc5.commit"}, + } + record = { + "container_id": "abc123", + "instance": target.instance, + "profile": target.profile, + "target": target.target, + } + api = mock.Mock(return_value={"status": "cancelled", "message": "newer request"}) + with ( + mock.patch.object(rvctl, "resolve_mcp_record", return_value=record), + mock.patch.object(rvctl, "validated_mcp_record"), + mock.patch.object(rvctl, "validate_mcp_card_mount"), + mock.patch.object(rvctl, "select_mcp_instance"), + mock.patch.object( + rvctl, + "verify_card_api_identity", + return_value={ + "id": "card-1", + "uuid": "uuid-1", + "version": "v1", + "source_sha256": "abc123", + }, + ), + mock.patch.object(rvctl, "card_api_json", api), + self.assertRaisesRegex(SystemExit, "did not reach ready: cancelled"), + ): + rvctl.select_session_stage(SimpleNamespace(), target, args, True, entry) + self.assertEqual(api.call_count, 1) + payload = api.call_args.kwargs["payload"] + self.assertEqual(payload["expected_identity"]["source_sha256"], "abc123") + + def test_mcp_registry_profile_and_target_must_match_container_labels(self) -> None: + record = { + "runtime": "podman", + "container_name": "container", + "container_id": "abc123", + "instance_key": "instance-key", + "profile": "hazard3-sim", + "target": "hazard3-baremetal", + "socket_dir": "/tmp/not-used", + } + inspection = [{ + "Id": "abc123", + "Config": {"Labels": { + "edu.stem.instance-key": "instance-key", + "edu.stem.profile": "hazard3-sim", + "edu.stem.target": "rp2350-rv", + }}, + }] + completed = subprocess.CompletedProcess([], 0, stdout=json.dumps(inspection), stderr="") + with ( + mock.patch.object(rvctl.shutil, "which", return_value="/usr/bin/podman"), + mock.patch.object(subprocess, "run", return_value=completed), + self.assertRaisesRegex(SystemExit, "Stale MCP registry"), + ): + rvctl.validated_mcp_record(record) + + def test_session_task_matching_is_anchored(self) -> None: + self.assertTrue(rvctl.same_task("task04", "task04_address_arithmetic_alloc")) + self.assertFalse(rvctl.same_task("notask04", "task04_address_arithmetic_alloc")) + self.assertFalse(rvctl.same_task("figure2026", "task04_address_arithmetic_alloc")) + if __name__ == "__main__": unittest.main()