feat(vault): ID scheme v2 + vault ls/show/chain browse commands (#1428)

feat(vault): ID scheme v2 + vault ls/show/chain browse commands

Author: profvjreddi

interviews/vault-cli/src/vault_cli/commands/authoring.py

@@ -38,13 +38,66 @@ def _short_hash(title: str) -> str:
     return hashlib.sha256(title.encode("utf-8")).hexdigest()[:6]
 
 
+def _id_hash(title: str, topic: str) -> str:
+    """ID scheme v2 hash: first 4 hex chars of sha256(title + "\\n" + topic).
+
+    Matches the recipe documented in interviews/vault/docs/ID_SCHEMES.md.
+    Including the topic defends against two unrelated questions with
+    identical titles hashing identically.
+    """
+    payload = f"{title}\n{topic}".encode("utf-8")
+    return hashlib.sha256(payload).hexdigest()[:4]
+
+
+def _yyyymm() -> str:
+    """Current year-month as 6 digits, for the v2 ID scheme."""
+    return datetime.now(UTC).strftime("%Y%m")
+
+
 def _slug(s: str) -> str:
     keep = "".join(c if c.isalnum() or c == "-" else "-" for c in s.lower())
     while "--" in keep:
         keep = keep.replace("--", "-")
     return keep.strip("-") or "untitled"
 
 
+def _new_question_id(track: str, title: str, topic: str, existing_ids: set[str]) -> tuple[str, str]:
+    """Mint a v2 question ID: <track>-<yyyymm>-<4hex>.
+
+    On collision within the same (track, yyyymm) bucket, increment the
+    4-hex suffix to the next free hex until a free slot is found.
+    Returns (id, hex_used).
+    """
+    yyyymm = _yyyymm()
+    base_hex = _id_hash(title, topic)
+    # Increment hex on collision (65,536 slots per bucket; collisions rare).
+    n = int(base_hex, 16)
+    for _ in range(0x10000):
+        candidate_hex = f"{n:04x}"
+        qid = f"{track}-{yyyymm}-{candidate_hex}"
+        if qid not in existing_ids:
+            return qid, candidate_hex
+        n = (n + 1) & 0xFFFF
+    raise RuntimeError(f"ID-space exhausted for bucket {track}-{yyyymm}")
+
+
+def _new_chain_id(track: str, topic: str, existing_ids: set[str]) -> str:
+    """Mint a v2 chain ID: chain-<track>-<topic-slug>-<yyyymm>[-<suffix>].
+
+    On collision add a single-letter suffix (a, b, c, …) to disambiguate.
+    """
+    yyyymm = _yyyymm()
+    slug = _slug(topic)
+    base = f"chain-{track}-{slug}-{yyyymm}"
+    if base not in existing_ids:
+        return base
+    for letter in "abcdefghijklmnopqrstuvwxyz":
+        candidate = f"{base}-{letter}"
+        if candidate not in existing_ids:
+            return candidate
+    raise RuntimeError(f"Chain-ID space exhausted for bucket {base}")
+
+
 def _git_user_email() -> str | None:
     """Resolve the committer identity for auto-populating ``authors:`` (David H4)."""
     try:
@@ -136,8 +189,7 @@ def new_cmd(
         rate, per §3.3 concurrency contract.
         """
         # v1.0: classification lives in YAML, filesystem uses track only.
-        topic_slug = _slug(topic)
-        h = _short_hash(title)
+        # v2 ID scheme: <track>-<yyyymm>-<4hex> (see docs/ID_SCHEMES.md).
 
         # §3.3: git pull --rebase on the registry before allocation.
         if not skip_rebase:
@@ -149,18 +201,14 @@ def new_cmd(
                     check=False, capture_output=True,
                 )
 
-        # Allocate seq by scanning existing files under the track directory.
+        # Build the set of existing IDs so _new_question_id can avoid collisions.
         cell_dir = path_for_question(vault_dir, track.value, "").parent
         cell_dir.mkdir(parents=True, exist_ok=True)
-        seq = 1
-        while True:
-            filename = f"{track.value}-{topic_slug}-{h}-{seq:04d}.yaml"
-            candidate = cell_dir / filename
-            if not candidate.exists():
-                break
-            seq += 1
+        existing_ids: set[str] = set()
+        for p in cell_dir.glob("*.yaml"):
+            existing_ids.add(p.stem)
 
-        qid = f"{track.value}-{topic_slug}-{h}-{seq:04d}"
+        qid, _ = _new_question_id(track.value, title, topic, existing_ids)
         now = _now()
 
         author = _git_user_email()

interviews/vault-cli/src/vault_cli/commands/chain.py

@@ -0,0 +1,120 @@
+"""``vault chain`` — browse and inspect question chains.
+
+Subcommands:
+    vault chain ls [--track --topic]          list chains with counts + spans
+    vault chain show <chain-id>                walk a chain end-to-end
+
+A chain links questions on a single topic into a progression, usually
+across Bloom's levels. ≈32% of the corpus participates in ≥1 chain;
+≈101 questions are in multiple chains.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from vault_cli.loader import load_all
+
+
+chain_app = typer.Typer(help="Browse and inspect question chains.")
+
+
+def _collect_chains(loaded) -> dict[str, list]:
+    """Return {chain_id: [(position, loaded_question), ...]}."""
+    out: dict[str, list] = {}
+    for lq in loaded:
+        for c in (lq.question.chains or []):
+            out.setdefault(c.id, []).append((c.position, lq))
+    for cid in out:
+        out[cid].sort(key=lambda x: x[0])
+    return out
+
+
+@chain_app.command("ls")
+def chain_ls(
+    track: str | None = typer.Option(None, "--track", help="Filter by track (cloud/edge/mobile/tinyml/global)."),
+    topic: str | None = typer.Option(None, "--topic", help="Filter by topic slug."),
+    vault_dir: Path = typer.Option(Path("interviews/vault"), "--vault-dir"),
+) -> None:
+    """List chains with member count + level span."""
+    console = Console()
+    loaded, _ = load_all(vault_dir)
+    chains = _collect_chains(loaded)
+
+    rows = []
+    for cid, members in chains.items():
+        topics = {m[1].question.topic for m in members}
+        tracks = {m[1].question.track for m in members}
+        levels = sorted({m[1].question.level for m in members})
+        first_topic = next(iter(topics))
+        first_track = next(iter(tracks))
+        if track and first_track != track:
+            continue
+        if topic and topic not in topics:
+            continue
+        rows.append((cid, first_track, first_topic, len(members), "/".join(levels)))
+
+    rows.sort(key=lambda r: (r[1], r[2], r[0]))
+
+    table = Table(show_header=True, header_style="bold")
+    table.add_column("chain_id", no_wrap=True, style="cyan")
+    table.add_column("track", no_wrap=True)
+    table.add_column("topic", no_wrap=True, style="dim")
+    table.add_column("#", justify="right")
+    table.add_column("level span")
+    for cid, tr, tp, n, sp in rows:
+        table.add_row(cid, tr, tp, str(n), sp)
+    console.print(table)
+    console.print(f"[dim]{len(rows)} chains[/dim]")
+
+
+@chain_app.command("show")
+def chain_show(
+    chain_id: str = typer.Argument(..., help="Chain ID to walk (e.g. cloud-chain-432)."),
+    vault_dir: Path = typer.Option(Path("interviews/vault"), "--vault-dir"),
+) -> None:
+    """Walk a chain end-to-end: one row per member, ordered by position."""
+    console = Console()
+    loaded, _ = load_all(vault_dir)
+    chains = _collect_chains(loaded)
+
+    members = chains.get(chain_id)
+    if members is None:
+        console.print(f"[red]chain not found:[/red] {chain_id!r}")
+        raise typer.Exit(code=1)
+
+    # Detect topic/track drift within the chain.
+    topics = sorted({m[1].question.topic for m in members})
+    tracks = sorted({m[1].question.track for m in members})
+    levels = [m[1].question.level for m in members]
+    levels_sorted = sorted(levels, key=lambda L: ("L1","L2","L3","L4","L5","L6+").index(L))
+    monotonic = levels == levels_sorted
+
+    console.print(f"[bold cyan]{chain_id}[/bold cyan]   "
+                  f"{len(members)} members   track(s)={','.join(tracks)}   topic(s)={','.join(topics)}")
+    if len(topics) > 1:
+        console.print(f"  [yellow]warning[/yellow]: chain spans multiple topics — likely mis-linked")
+    if not monotonic:
+        console.print(f"  [yellow]warning[/yellow]: levels not monotonically non-decreasing across positions")
+
+    table = Table(show_header=True, header_style="bold")
+    table.add_column("#", justify="right", no_wrap=True)
+    table.add_column("level", no_wrap=True)
+    table.add_column("zone", no_wrap=True)
+    table.add_column("id", no_wrap=True, style="cyan")
+    table.add_column("title")
+    for pos, lq in members:
+        q = lq.question
+        table.add_row(str(pos), q.level, q.zone, q.id, q.title)
+    console.print(table)
+
+
+def register(app: typer.Typer) -> None:
+    app.add_typer(chain_app, name="chain")
+
+
+__all__ = ["register"]

interviews/vault-cli/src/vault_cli/commands/ls.py

@@ -0,0 +1,89 @@
+"""``vault ls`` — browse questions with axis filters.
+
+One-line-per-question output, aligned columns, filterable on every
+first-class classification axis. Powers the "I want to see level at a
+glance" workflow without opening individual YAMLs.
+
+Usage:
+    vault ls                             # every question in the vault
+    vault ls --track cloud               # cloud-only
+    vault ls --level L6+                 # only L6+ questions
+    vault ls --zone mastery              # only mastery-zone
+    vault ls --topic kv-cache-management
+    vault ls --status published
+    vault ls --in-chains                 # only questions that are in >=1 chain
+    vault ls --track cloud --level L4 --zone diagnosis   # combinable
+
+Output columns: id | track | level | zone | topic | #chains | title
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from vault_cli.loader import load_all
+
+
+def register(app: typer.Typer) -> None:
+    @app.command("ls")
+    def ls_cmd(
+        track: str | None = typer.Option(None, "--track", help="Filter by track."),
+        level: str | None = typer.Option(None, "--level", help="Filter by level (L1 … L6+)."),
+        zone: str | None = typer.Option(None, "--zone", help="Filter by zone (one of the 11 ikigai zones)."),
+        topic: str | None = typer.Option(None, "--topic", help="Filter by topic slug."),
+        status: str | None = typer.Option(None, "--status", help="Filter by status (published, draft, flagged, archived, deleted)."),
+        in_chains: bool = typer.Option(False, "--in-chains", help="Only questions that belong to ≥1 chain."),
+        vault_dir: Path = typer.Option(
+            Path("interviews/vault"), "--vault-dir",
+            help="Vault root.",
+        ),
+        limit: int = typer.Option(0, "--limit", "-n", help="Truncate output after N rows (0 = all)."),
+        plain: bool = typer.Option(False, "--plain", help="Plain tab-separated output (for piping to awk/grep)."),
+    ) -> None:
+        """List questions with axis filters. Aligned output or TSV."""
+        console = Console()
+        loaded, errors = load_all(vault_dir)
+        if errors:
+            console.print(f"[yellow]warning[/yellow]: {len(errors)} load errors (skipped)")
+
+        rows = []
+        for lq in loaded:
+            q = lq.question
+            if track and q.track != track: continue
+            if level and q.level != level: continue
+            if zone and q.zone != zone: continue
+            if topic and q.topic != topic: continue
+            if status and q.status != status: continue
+            n_chains = len(q.chains or [])
+            if in_chains and n_chains == 0: continue
+            rows.append((q.id, q.track, q.level, q.zone, q.topic, n_chains, q.title))
+
+        rows.sort(key=lambda r: (r[1], r[2], r[0]))   # track, level, id
+        if limit:
+            rows = rows[:limit]
+
+        if plain:
+            for r in rows:
+                print("\t".join(str(x) for x in r))
+            console.print(f"[dim]({len(rows)} rows)[/dim]", style="dim")
+            return
+
+        table = Table(show_header=True, header_style="bold")
+        table.add_column("id", no_wrap=True, style="cyan")
+        table.add_column("track", no_wrap=True)
+        table.add_column("level", no_wrap=True)
+        table.add_column("zone", no_wrap=True)
+        table.add_column("topic", no_wrap=True, style="dim")
+        table.add_column("ch", justify="right", style="dim")
+        table.add_column("title", no_wrap=False)
+        for qid, tr, lv, zn, tp, nc, ti in rows:
+            table.add_row(qid, tr, lv, zn, tp, str(nc), ti)
+        console.print(table)
+        console.print(f"[dim]{len(rows)} rows[/dim]")
+
+
+__all__ = ["register"]