Add slash commands (/status, /list, /lint, /add) to chat REPL

Extract core logic from CLI commands into reusable functions
(print_status, print_list, run_lint) so the chat REPL can call
them directly instead of duplicating code. /add uses
asyncio.to_thread since _add_single_file internally calls
asyncio.run which cannot be nested.

Author: KylinMountain

openkb/agent/chat.py

@@ -46,6 +46,10 @@
     "  /exit          Exit (Ctrl-D also works)\n"
     "  /clear         Start a fresh session (current one is kept on disk)\n"
     "  /save [name]   Export transcript to wiki/explorations/\n"
+    "  /status        Show knowledge base status\n"
+    "  /list          List all documents in the knowledge base\n"
+    "  /lint          Lint the knowledge base\n"
+    "  /add <path>    Add a document or directory to the knowledge base\n"
     "  /help          Show this"
 )
 
@@ -271,6 +275,37 @@ def _save_transcript(kb_dir: Path, session: ChatSession, name: str | None) -> Pa
     return path
 
 
+async def _run_add(arg: str, kb_dir: Path, style: Style) -> None:
+    """Add a document or directory to the knowledge base from the chat REPL."""
+    import asyncio
+    from openkb.cli import _add_single_file, SUPPORTED_EXTENSIONS
+
+    target = Path(arg).expanduser()
+    if not target.is_absolute():
+        target = Path.cwd() / target
+    target = target.resolve()
+
+    if not target.exists():
+        _fmt(style, ("class:error", f"Path does not exist: {arg}\n"))
+        return
+
+    if target.is_dir():
+        files = [
+            f for f in sorted(target.rglob("*"))
+            if f.is_file() and f.suffix.lower() in SUPPORTED_EXTENSIONS
+        ]
+        if not files:
+            _fmt(style, ("class:error", f"No supported files found in {arg}.\n"))
+            return
+        for f in files:
+            await asyncio.to_thread(_add_single_file, f, kb_dir)
+    else:
+        if target.suffix.lower() not in SUPPORTED_EXTENSIONS:
+            _fmt(style, ("class:error", f"Unsupported file type: {target.suffix}\n"))
+            return
+        await asyncio.to_thread(_add_single_file, target, kb_dir)
+
+
 async def _handle_slash(
     cmd: str,
     kb_dir: Path,
@@ -307,6 +342,28 @@ async def _handle_slash(
         _fmt(style, ("class:slash.ok", f"Saved to {path}\n"))
         return None
 
+    if head == "/status":
+        from openkb.cli import print_status
+        print_status(kb_dir)
+        return None
+
+    if head == "/list":
+        from openkb.cli import print_list
+        print_list(kb_dir)
+        return None
+
+    if head == "/lint":
+        from openkb.cli import run_lint
+        await run_lint(kb_dir)
+        return None
+
+    if head == "/add":
+        if not arg:
+            _fmt(style, ("class:error", "Usage: /add <path>\n"))
+            return None
+        await _run_add(arg, kb_dir, style)
+        return None
+
     _fmt(
         style,
         ("class:error", f"Unknown command: {head}. Try /help.\n"),

openkb/cli.py

@@ -525,18 +525,12 @@ def on_new_files(paths):
     watch_directory(raw_dir, on_new_files)
 
 
-@cli.command()
-@click.option("--fix", is_flag=True, default=False, help="Automatically fix lint issues (not yet implemented).")
-@click.pass_context
-def lint(ctx, fix):
-    """Lint the knowledge base for structural and semantic inconsistencies."""
-    if fix:
-        click.echo("Warning: --fix is not yet implemented. Running lint in report-only mode.")
-    kb_dir = _find_kb_dir(ctx.obj.get("kb_dir_override"))
-    if kb_dir is None:
-        click.echo("No knowledge base found. Run `openkb init` first.")
-        return
+async def run_lint(kb_dir: Path) -> Path:
+    """Run structural + knowledge lint, write report, return report path.
 
+    Async because knowledge lint uses an LLM agent. Usable from CLI
+    (via ``asyncio.run``) and directly from the chat REPL.
+    """
     from openkb.lint import run_structural_lint
     from openkb.agent.linter import run_knowledge_lint
 
@@ -556,15 +550,13 @@ def lint(ctx, fix):
     _setup_llm_key(kb_dir)
     model: str = config.get("model", DEFAULT_CONFIG["model"])
 
-    # Structural lint
     click.echo("Running structural lint...")
     structural_report = run_structural_lint(kb_dir)
     click.echo(structural_report)
 
-    # Knowledge lint (semantic)
     click.echo("Running knowledge lint...")
     try:
-        knowledge_report = asyncio.run(run_knowledge_lint(kb_dir, model))
+        knowledge_report = await run_knowledge_lint(kb_dir, model)
     except Exception as exc:
         knowledge_report = f"Knowledge lint failed: {exc}"
     click.echo(knowledge_report)
@@ -579,17 +571,25 @@ def lint(ctx, fix):
     report_path.write_text(report_content, encoding="utf-8")
     append_log(kb_dir / "wiki", "lint", f"report → {report_path.name}")
     click.echo(f"\nReport written to {report_path}")
+    return report_path
 
 
-@cli.command(name="list")
+@cli.command()
+@click.option("--fix", is_flag=True, default=False, help="Automatically fix lint issues (not yet implemented).")
 @click.pass_context
-def list_cmd(ctx):
-    """List all documents in the knowledge base."""
+def lint(ctx, fix):
+    """Lint the knowledge base for structural and semantic inconsistencies."""
+    if fix:
+        click.echo("Warning: --fix is not yet implemented. Running lint in report-only mode.")
     kb_dir = _find_kb_dir(ctx.obj.get("kb_dir_override"))
     if kb_dir is None:
         click.echo("No knowledge base found. Run `openkb init` first.")
         return
+    asyncio.run(run_lint(kb_dir))
+
 
+def print_list(kb_dir: Path) -> None:
+    """Print all documents in the knowledge base. Usable from CLI and chat REPL."""
     openkb_dir = kb_dir / ".openkb"
     hashes_file = openkb_dir / "hashes.json"
     if not hashes_file.exists():
@@ -642,15 +642,19 @@ def list_cmd(ctx):
                 click.echo(f"  - {r}")
 
 
-@cli.command()
+@cli.command(name="list")
 @click.pass_context
-def status(ctx):
-    """Show the current status of the knowledge base."""
+def list_cmd(ctx):
+    """List all documents in the knowledge base."""
     kb_dir = _find_kb_dir(ctx.obj.get("kb_dir_override"))
     if kb_dir is None:
         click.echo("No knowledge base found. Run `openkb init` first.")
         return
+    print_list(kb_dir)
 
+
+def print_status(kb_dir: Path) -> None:
+    """Print knowledge base status. Usable from CLI and chat REPL."""
     wiki_dir = kb_dir / "wiki"
     subdirs = ["sources", "summaries", "concepts", "reports"]
 
@@ -698,3 +702,14 @@ def status(ctx):
             import datetime
             mtime = datetime.datetime.fromtimestamp(newest_report.stat().st_mtime)
             click.echo(f"  Last lint:     {mtime.strftime('%Y-%m-%d %H:%M:%S')}")
+
+
+@cli.command()
+@click.pass_context
+def status(ctx):
+    """Show the current status of the knowledge base."""
+    kb_dir = _find_kb_dir(ctx.obj.get("kb_dir_override"))
+    if kb_dir is None:
+        click.echo("No knowledge base found. Run `openkb init` first.")
+        return
+    print_status(kb_dir)