refactor(cli): start the command registry, migrate merge-chunks (#1212)

graphify/cli.py

@@ -712,6 +712,13 @@ def _reenter_main() -> None:
 
 
 def dispatch_command(cmd: str) -> None:
+    from graphify.commands import COMMANDS
+
+    handler = COMMANDS.get(cmd)
+    if handler is not None:
+        handler()
+        return
+
     if cmd == "provider":
         from graphify.llm import _custom_providers_path, BACKENDS
         import json as _json
@@ -3697,87 +3704,6 @@ def _invalidate_file_manifest_for_db_graph() -> None:
         (out / ".graphify_uncached.txt").write_text("\n".join(uncached), encoding="utf-8")
         print(f"Cache: {len(files) - len(uncached)} hit, {len(uncached)} miss")
 
-    elif cmd == "merge-chunks":
-        # graphify merge-chunks <chunk_glob_or_files...> --out <path>
-        # Concatenates .graphify_chunk_*.json files written by semantic subagents.
-        # Deduplicates nodes by id (first writer wins). Sums token counts.
-        import glob as _glob
-        if len(sys.argv) < 3:
-            print("Usage: graphify merge-chunks <chunk_files...> --out <path>", file=sys.stderr)
-            sys.exit(1)
-        out_path: Path | None = None
-        chunk_args: list[str] = []
-        i = 2
-        while i < len(sys.argv):
-            if sys.argv[i] == "--out" and i + 1 < len(sys.argv):
-                out_path = Path(sys.argv[i + 1])
-                i += 2
-            else:
-                chunk_args.append(sys.argv[i])
-                i += 1
-        if not out_path:
-            print("error: --out <path> required", file=sys.stderr)
-            sys.exit(1)
-        chunk_files: list[str] = []
-        for arg in chunk_args:
-            expanded = _glob.glob(arg)
-            chunk_files.extend(sorted(expanded) if expanded else [arg])
-        merged: dict = {"nodes": [], "edges": [], "hyperedges": [], "input_tokens": 0, "output_tokens": 0}
-        seen_ids: set[str] = set()
-        valid_chunks = 0
-        # These chunk files are untrusted subagent output. load_validated_...
-        # stats the file size BEFORE reading it (so a multi-GB chunk can't blow up
-        # memory), parses the JSON, and validates the security caps + the node/
-        # edge id charset that blocks path traversal (#825) — the same enforcement
-        # the skill merge path applies. A bad chunk is skipped with a warning
-        # while valid siblings still merge; if every chunk is invalid, fail
-        # closed instead of reporting success and replacing --out with an empty
-        # semantic layer. Deliberately NOT wired into
-        # build_from_json/load_graph_json, which must keep loading valid
-        # pre-existing graphs. file_type is left to build's coercion (#840).
-        from graphify.semantic_cleanup import load_validated_semantic_fragment
-        for cf in chunk_files:
-            chunk, _chunk_errs = load_validated_semantic_fragment(Path(cf))
-            if _chunk_errs:
-                print(
-                    f"[graphify merge-chunks] warning: skipping invalid chunk {cf}: "
-                    f"{'; '.join(_chunk_errs[:3])}",
-                    file=sys.stderr,
-                )
-                continue
-            valid_chunks += 1
-            for n in chunk.get("nodes", []):
-                if n.get("id") not in seen_ids:
-                    seen_ids.add(n["id"])
-                    merged["nodes"].append(n)
-            merged["edges"].extend(chunk.get("edges", []))
-            merged["hyperedges"].extend(chunk.get("hyperedges", []))
-            # Coerce token counts: a chunk is untrusted, so a non-numeric
-            # input_tokens/output_tokens must not abort the whole merge with a
-            # TypeError after other chunks already merged.
-            for _tok in ("input_tokens", "output_tokens"):
-                _v = chunk.get(_tok, 0)
-                merged[_tok] += _v if isinstance(_v, (int, float)) else 0
-        if not valid_chunks:
-            print(
-                f"[graphify merge-chunks] error: no valid chunks to merge; "
-                f"refusing to write {out_path}",
-                file=sys.stderr,
-            )
-            sys.exit(1)
-        out_path.parent.mkdir(parents=True, exist_ok=True)
-        from graphify.paths import write_json_atomic as _wja
-        _wja(out_path, merged, ensure_ascii=False)
-        chunk_summary = (
-            f"{valid_chunks} chunks"
-            if valid_chunks == len(chunk_files)
-            else f"{valid_chunks} of {len(chunk_files)} chunks"
-        )
-        print(
-            f"Merged {chunk_summary}: {len(merged['nodes'])} nodes, {len(merged['edges'])} edges, "
-            f"{merged['input_tokens']:,} in / {merged['output_tokens']:,} out tokens"
-        )
-
     elif cmd == "merge-semantic":
         # graphify merge-semantic --cached <path> --new <path> --out <path>
         # Merges cached semantic results with freshly-extracted chunk results.

graphify/commands/MIGRATION.md

@@ -0,0 +1,42 @@
+# Migrating a command out of cli.dispatch_command
+
+`graphify/cli.py`'s `dispatch_command()` is a ~3,100-line `if/elif` chain over
+the command name (issue #1212 proposes splitting it into this package, mirroring
+the `graphify/extractors/` split). This is the playbook for porting ONE command.
+It is written so an AI agent can execute it in a single session.
+
+## Status
+
+| command | migrated |
+|---|---|
+| merge-chunks | yes |
+| (everything else in `dispatch_command`) | no |
+
+## Invariants (non-negotiable)
+
+1. **Verbatim moves only.** Move the body of one `elif cmd == "<name>":` branch
+   into a function, dedented one level, with no renames, no docstring edits, no
+   reformatting, no added annotations, no "improvements". The handler reads
+   `sys.argv` and calls `sys.exit(...)` exactly as the branch did, so the move
+   is behavior-preserving. Verify: the command's existing tests pass unchanged.
+2. **One command per PR.** Small diffs keep review trivial and avoid conflicts
+   with other in-flight ports.
+3. **Registry-first dispatch, if/elif fallback.** `dispatch_command` consults
+   `COMMANDS` before its remaining chain, so a migrated command must be *removed*
+   from the `if/elif` chain in the same PR. `test_commands_registry.py` fails if
+   a name is both registered and branched (a dead old body).
+
+## Steps
+
+1. Pick a command that does not share local state with its neighbours (most
+   don't — each branch reads `sys.argv` and exits). Note its existing test
+   (e.g. `merge-chunks` → `tests/test_merge_chunks_validation.py`).
+2. Create `graphify/commands/<name>.py` with `def <name>() -> None:` and paste
+   the branch body verbatim, dedented one level. Keep the branch's local imports
+   inside the function.
+3. Register it in `graphify/commands/__init__.py`: import the function and add
+   `"<command-name>": <name>` to `COMMANDS`.
+4. Delete the `elif cmd == "<command-name>":` branch (whole block) from
+   `dispatch_command` in `cli.py`.
+5. Run the command's existing test plus `tests/test_commands_registry.py`. Both
+   must pass. Update the status table above.

graphify/commands/__init__.py

@@ -0,0 +1,17 @@
+"""Per-command handlers, incrementally migrated out of cli.dispatch_command.
+
+Dispatch still flows through graphify.cli.dispatch_command, which consults
+COMMANDS first and falls through to the remaining if/elif chain for commands
+that have not been ported yet. Each handler reads sys.argv and calls sys.exit
+exactly as its original branch did, so a move is behavior-preserving. See
+MIGRATION.md for how to port another command. Tracks issue #1212.
+"""
+from __future__ import annotations
+
+from typing import Callable
+
+from graphify.commands.merge_chunks import merge_chunks
+
+COMMANDS: dict[str, Callable[[], None]] = {
+    "merge-chunks": merge_chunks,
+}

graphify/commands/merge_chunks.py

@@ -0,0 +1,92 @@
+"""`graphify merge-chunks` — concatenate semantic subagent chunk files.
+
+Moved verbatim from the merge-chunks branch of cli.dispatch_command as the
+first step of the __main__/cli command split (issue #1212). See
+graphify/commands/MIGRATION.md.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+
+def merge_chunks() -> None:
+    # graphify merge-chunks <chunk_glob_or_files...> --out <path>
+    # Concatenates .graphify_chunk_*.json files written by semantic subagents.
+    # Deduplicates nodes by id (first writer wins). Sums token counts.
+    import glob as _glob
+    if len(sys.argv) < 3:
+        print("Usage: graphify merge-chunks <chunk_files...> --out <path>", file=sys.stderr)
+        sys.exit(1)
+    out_path: Path | None = None
+    chunk_args: list[str] = []
+    i = 2
+    while i < len(sys.argv):
+        if sys.argv[i] == "--out" and i + 1 < len(sys.argv):
+            out_path = Path(sys.argv[i + 1])
+            i += 2
+        else:
+            chunk_args.append(sys.argv[i])
+            i += 1
+    if not out_path:
+        print("error: --out <path> required", file=sys.stderr)
+        sys.exit(1)
+    chunk_files: list[str] = []
+    for arg in chunk_args:
+        expanded = _glob.glob(arg)
+        chunk_files.extend(sorted(expanded) if expanded else [arg])
+    merged: dict = {"nodes": [], "edges": [], "hyperedges": [], "input_tokens": 0, "output_tokens": 0}
+    seen_ids: set[str] = set()
+    valid_chunks = 0
+    # These chunk files are untrusted subagent output. load_validated_...
+    # stats the file size BEFORE reading it (so a multi-GB chunk can't blow up
+    # memory), parses the JSON, and validates the security caps + the node/
+    # edge id charset that blocks path traversal (#825) — the same enforcement
+    # the skill merge path applies. A bad chunk is skipped with a warning
+    # while valid siblings still merge; if every chunk is invalid, fail
+    # closed instead of reporting success and replacing --out with an empty
+    # semantic layer. Deliberately NOT wired into
+    # build_from_json/load_graph_json, which must keep loading valid
+    # pre-existing graphs. file_type is left to build's coercion (#840).
+    from graphify.semantic_cleanup import load_validated_semantic_fragment
+    for cf in chunk_files:
+        chunk, _chunk_errs = load_validated_semantic_fragment(Path(cf))
+        if _chunk_errs:
+            print(
+                f"[graphify merge-chunks] warning: skipping invalid chunk {cf}: "
+                f"{'; '.join(_chunk_errs[:3])}",
+                file=sys.stderr,
+            )
+            continue
+        valid_chunks += 1
+        for n in chunk.get("nodes", []):
+            if n.get("id") not in seen_ids:
+                seen_ids.add(n["id"])
+                merged["nodes"].append(n)
+        merged["edges"].extend(chunk.get("edges", []))
+        merged["hyperedges"].extend(chunk.get("hyperedges", []))
+        # Coerce token counts: a chunk is untrusted, so a non-numeric
+        # input_tokens/output_tokens must not abort the whole merge with a
+        # TypeError after other chunks already merged.
+        for _tok in ("input_tokens", "output_tokens"):
+            _v = chunk.get(_tok, 0)
+            merged[_tok] += _v if isinstance(_v, (int, float)) else 0
+    if not valid_chunks:
+        print(
+            f"[graphify merge-chunks] error: no valid chunks to merge; "
+            f"refusing to write {out_path}",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    from graphify.paths import write_json_atomic as _wja
+    _wja(out_path, merged, ensure_ascii=False)
+    chunk_summary = (
+        f"{valid_chunks} chunks"
+        if valid_chunks == len(chunk_files)
+        else f"{valid_chunks} of {len(chunk_files)} chunks"
+    )
+    print(
+        f"Merged {chunk_summary}: {len(merged['nodes'])} nodes, {len(merged['edges'])} edges, "
+        f"{merged['input_tokens']:,} in / {merged['output_tokens']:,} out tokens"
+    )

tests/test_commands_registry.py

@@ -0,0 +1,30 @@
+"""The command registry must dispatch, and its keys must not collide with the
+if/elif branches still living in cli.dispatch_command.
+
+Part of the __main__/cli command split (issue #1212): commands migrate one at
+a time into graphify/commands/, and dispatch_command consults COMMANDS before
+its remaining if/elif chain. A key that also appears as an `elif cmd == "..."`
+branch would mean a half-migrated command whose old body is now dead — this
+test fails if that ever happens.
+"""
+import re
+from pathlib import Path
+
+from graphify.commands import COMMANDS
+
+CLI_SOURCE = (Path(__file__).resolve().parent.parent / "graphify" / "cli.py").read_text(encoding="utf-8")
+BRANCH_COMMANDS = set(re.findall(r'cmd == "([^"]+)"', CLI_SOURCE))
+
+
+def test_registry_is_populated_and_callable() -> None:
+    assert COMMANDS, "COMMANDS registry is empty"
+    for name, handler in COMMANDS.items():
+        assert callable(handler), f"handler for {name!r} is not callable"
+
+
+def test_no_command_is_both_registered_and_branched() -> None:
+    overlap = sorted(set(COMMANDS) & BRANCH_COMMANDS)
+    assert overlap == [], (
+        f"these commands are in COMMANDS but still have an if/elif branch in "
+        f"cli.dispatch_command (dead branch): {overlap}"
+    )