feat(cli): add /diff slash command to show git changes in working directory

cli.py

@@ -4750,6 +4750,44 @@ def _handle_copy_command(self, cmd_original: str) -> None:
         except Exception as e:
             _cprint(f"  Clipboard copy failed: {e}")
 
+    def _handle_diff_command(self, cmd_original: str) -> None:
+        """Handle /diff [args...] — show git diff in the working directory."""
+        parts = cmd_original.split(maxsplit=1)
+        extra_args = parts[1].strip() if len(parts) > 1 else ""
+
+        if not shutil.which("git"):
+            _cprint("  git is not installed or not on PATH.")
+            return
+
+        try:
+            subprocess.run(
+                ["git", "rev-parse", "--git-dir"],
+                capture_output=True, check=True,
+                cwd=self.cwd if hasattr(self, "cwd") else None,
+            )
+        except subprocess.CalledProcessError:
+            _cprint("  Not in a git repository (or any parent up to root).")
+            return
+
+        cmd = ["git", "diff"]
+        if extra_args:
+            cmd.extend(extra_args.split())
+
+        # Pipe through less if output is a TTY and no arguments were given
+        # (preserving git's colour output).
+        env = dict(os.environ, GIT_PAGER="cat")
+        if sys.stdout.isatty() and not extra_args:
+            env.pop("GIT_PAGER", None)
+
+        try:
+            subprocess.run(
+                cmd,
+                cwd=self.cwd if hasattr(self, "cwd") else None,
+                env=env,
+            )
+        except Exception as e:
+            _cprint(f"  Failed to run git diff: {e}")
+
     def _handle_image_command(self, cmd_original: str):
         """Handle /image <path> — attach a local image file for the next prompt."""
         raw_args = (cmd_original.split(None, 1)[1].strip() if " " in cmd_original else "")
@@ -6930,6 +6968,8 @@ def process_command(self, command: str) -> bool:
             self._show_insights(cmd_original)
         elif canonical == "copy":
             self._handle_copy_command(cmd_original)
+        elif canonical == "diff":
+            self._handle_diff_command(cmd_original)
         elif canonical == "debug":
             self._handle_debug_command()
         elif canonical == "paste":

hermes_cli/commands.py

@@ -193,6 +193,8 @@ class CommandDef:
                cli_only=True, aliases=("gateway",)),
     CommandDef("copy", "Copy the last assistant response to clipboard", "Info",
                cli_only=True, args_hint="[number]"),
+    CommandDef("diff", "Show git changes in the working directory", "Info",
+               cli_only=True, args_hint="[--staged] [path...]"),
     CommandDef("paste", "Attach clipboard image from your clipboard", "Info",
                cli_only=True),
     CommandDef("image", "Attach a local image file for your next prompt", "Info",

hermes_cli/completion.py

@@ -216,9 +216,9 @@ def generate_zsh(parser: argparse.ArgumentParser) -> str:
     typeset -A opt_args
 
     _arguments -C \\
-        '(-h --help){{-h,--help}}[Show help and exit]' \\
-        '(-V --version){{-V,--version}}[Show version and exit]' \\
-        '(-p --profile){{-p,--profile}}[Profile name]:profile:_hermes_profiles' \\
+        '(-)'{{-h,--help}}'[Show help and exit]' \\
+        '(-)'{{-V,--version}}'[Show version and exit]' \\
+        '(-)'{{-p,--profile}}'[Profile name]:profile:_hermes_profiles' \\
         '1:command:->commands' \\
         '*::arg:->args'
 

plugins/memory/holographic/store.py

@@ -175,6 +175,11 @@ def add_fact(
                 row = self._conn.execute(
                     "SELECT fact_id FROM facts WHERE content = ?", (content,)
                 ).fetchone()
+                if row is None:
+                    raise RuntimeError(
+                        f"Concurrent delete: fact with content '{content[:80]}' "
+                        "was removed between duplicate INSERT and SELECT"
+                    )
                 return int(row["fact_id"])
 
             # Entity extraction and linking
@@ -296,9 +301,12 @@ def update_fact(
             if content is not None:
                 self._compute_hrr_vector(fact_id, content)
             # Rebuild bank for relevant category
-            cat = category or self._conn.execute(
-                "SELECT category FROM facts WHERE fact_id = ?", (fact_id,)
-            ).fetchone()["category"]
+            cat = category
+            if cat is None:
+                row = self._conn.execute(
+                    "SELECT category FROM facts WHERE fact_id = ?", (fact_id,)
+                ).fetchone()
+                cat = row["category"] if row is not None else "general"
             self._rebuild_bank(cat)
 
             return True

tests/hermes_cli/test_completion.py

@@ -140,6 +140,38 @@ def test_nested_describe_blocks(self):
         # gateway has subcommands so a _cmds array must be generated
         assert "gateway_cmds" in out
 
+    def test_valid_zsh_arguments_syntax(self):
+        """_arguments entries must use valid zsh syntax.
+
+        The pattern '(-X --Y){-X,--Y}[...]' is invalid because () groups
+        cannot contain long options with spaces.  Correct form:
+        '(-)'{-h,--help}'[...]'
+        """
+        out = generate_zsh(_make_parser())
+        # Reject the broken pattern: a () group containing a long option
+        bad = re.findall(r"\('\([^)]*--\w+", out)
+        assert not bad, f"Invalid _arguments syntax found: {bad}"
+        # Verify the correct pattern is present for top-level options
+        assert "'(-)'{-h,--help}'" in out
+        assert "'(-)'{-V,--version}'" in out
+        assert "'(-)'{-p,--profile}'" in out
+
+    def test_valid_zsh_arguments_syntax(self):
+        """_arguments entries must use valid zsh syntax.
+
+        The pattern '(-X --Y){-X,--Y}[...]' is invalid because () groups
+        cannot contain long options with spaces.  Correct form:
+        '(-)'{-h,--help}'[...]'
+        """
+        out = generate_zsh(_make_parser())
+        # Reject the broken pattern: a () group containing a long option
+        bad = re.findall(r"\('\([^)]*--\w+", out)
+        assert not bad, f"Invalid _arguments syntax found: {bad}"
+        # Verify the correct pattern is present for top-level options
+        assert "'(-)'{-h,--help}'" in out
+        assert "'(-)'{-V,--version}'" in out
+        assert "'(-)'{-p,--profile}'" in out
+
 
 # ---------------------------------------------------------------------------
 # 4. Fish output

tests/plugins/memory/test_holographic_store_null_guards.py

@@ -0,0 +1,162 @@
+"""Tests for null-guard fix in plugins.memory.holographic.store.MemoryStore.
+
+Covers two concurrent-delete edge cases in add_fact() and update_fact() where
+fetchone() returns None:
+  - add_fact: IntegrityError → concurrent delete → RuntimeError, not TypeError
+  - update_fact: category re-fetch after concurrent delete → "general" fallback
+"""
+
+import sqlite3
+import tempfile
+
+import pytest
+
+from plugins.memory.holographic.store import MemoryStore
+
+
+@pytest.fixture
+def store():
+    """Create a MemoryStore backed by a temporary database."""
+    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
+        path = f.name
+    store = MemoryStore(db_path=path)
+    yield store
+    store._conn.close()
+
+
+# ---------------------------------------------------------------------------
+# Connection wrapper — lets us intercept .execute() calls despite
+# sqlite3.Connection being an immutable C extension type.
+# ---------------------------------------------------------------------------
+
+
+class _ExecuteWrapper:
+    """Proxy that wraps a sqlite3.Connection and intercepts .execute()."""
+
+    def __init__(self, real_conn):
+        object.__setattr__(self, "_conn", real_conn)
+        object.__setattr__(self, "_hooks", {})
+
+    def execute(self, sql, parameters=()):
+        hook = self._hooks.get("execute")
+        if hook is not None:
+            return hook(self._conn, sql, parameters)
+        return self._conn.execute(sql, parameters)
+
+    def __getattr__(self, name):
+        return getattr(self._conn, name)
+
+    def __setattr__(self, name, value):
+        if name in ("_conn", "_hooks"):
+            object.__setattr__(self, name, value)
+        else:
+            setattr(self._conn, name, value)
+
+
+# ---------------------------------------------------------------------------
+# add_fact tests
+# ---------------------------------------------------------------------------
+
+
+def test_add_fact_duplicate_returns_existing_id(store):
+    """Normal deduplication path: second add_fact with same content returns
+    the original fact_id without raising."""
+    fact_id_1 = store.add_fact("unique fact")
+    fact_id_2 = store.add_fact("unique fact")
+    assert fact_id_1 == fact_id_2
+    assert isinstance(fact_id_2, int)
+
+
+def test_add_fact_concurrent_delete_raises_runtime_error(store):
+    """When a concurrent process deletes the fact between the failed INSERT
+    and the follow-up SELECT, add_fact must raise RuntimeError with a
+    descriptive message — not a bare TypeError from subscripting None."""
+    store.add_fact("will be raced")
+
+    wrapper = _ExecuteWrapper(store._conn)
+    call_count = [0]
+
+    def interceptor(real_conn, sql, parameters):
+        call_count[0] += 1
+        if call_count[0] == 1:
+            raise sqlite3.IntegrityError("UNIQUE constraint failed: facts.content")
+        # Simulate concurrent delete: delete the row from a second
+        # connection so that the real SELECT returns no rows.
+        conn2 = sqlite3.connect(str(store.db_path))
+        conn2.execute("DELETE FROM facts WHERE content = ?", ("will be raced",))
+        conn2.commit()
+        conn2.close()
+        return real_conn.execute(sql, parameters)
+
+    wrapper._hooks["execute"] = interceptor
+    store._conn = wrapper
+
+    with pytest.raises(RuntimeError, match="Concurrent delete"):
+        store.add_fact("will be raced")
+
+
+# ---------------------------------------------------------------------------
+# update_fact tests
+# ---------------------------------------------------------------------------
+
+
+def test_update_fact_nonexistent_returns_false(store):
+    """Calling update_fact on a non-existent fact_id must return False."""
+    assert store.update_fact(99999, trust_delta=0.1) is False
+
+
+def test_update_fact_with_category_skips_refetch(store):
+    """When category is supplied by the caller, the re-fetch SELECT must
+    be skipped entirely."""
+    fact_id = store.add_fact("test content", category="work")
+
+    wrapper = _ExecuteWrapper(store._conn)
+    category_select_calls = []
+
+    def interceptor(real_conn, sql, parameters):
+        if isinstance(sql, str) and "SELECT category FROM facts" in sql:
+            category_select_calls.append(sql)
+        return real_conn.execute(sql, parameters)
+
+    wrapper._hooks["execute"] = interceptor
+    store._conn = wrapper
+
+    result = store.update_fact(fact_id, category="home", trust_delta=0.1)
+    assert result is True
+    assert len(category_select_calls) == 0, (
+        "category SELECT should be skipped when category is provided"
+    )
+
+
+def test_update_fact_concurrent_delete_uses_general_fallback(store):
+    """When category is NOT supplied and a concurrent delete removes the
+    fact between the UPDATE and the category re-fetch, the code must fall
+    back to 'general' instead of crashing with a TypeError."""
+    fact_id = store.add_fact("test content", category="work")
+
+    wrapper = _ExecuteWrapper(store._conn)
+    select_category_triggered = [False]
+    row_deleted = [False]
+
+    def interceptor(real_conn, sql, parameters):
+        if isinstance(sql, str) and "SELECT category FROM facts" in sql:
+            select_category_triggered[0] = True
+            # Delete the row from a second connection to simulate race,
+            # then run the real SELECT which will now return no rows.
+            if not row_deleted[0]:
+                conn2 = sqlite3.connect(str(store.db_path))
+                conn2.execute(
+                    "DELETE FROM facts WHERE fact_id = ?", (fact_id,)
+                )
+                conn2.commit()
+                conn2.close()
+                row_deleted[0] = True
+        return real_conn.execute(sql, parameters)
+
+    wrapper._hooks["execute"] = interceptor
+    store._conn = wrapper
+
+    # Must not crash — should fall back to "general"
+    result = store.update_fact(fact_id, trust_delta=0.1)
+    assert result is True
+    assert select_category_triggered[0], "category re-fetch was never triggered"