Refactor(core): split built-in command handlers

Author: kesimo

facio/command/builtin.py

@@ -0,0 +1 @@
+"""Built-in slash command handler modules."""

facio/command/builtins/__init__.py

@@ -0,0 +1,144 @@
+"""Guardrail slash command handlers."""
+
+from __future__ import annotations
+
+from facio.bus.events import OutboundMessage
+from facio.command.router import CommandContext
+
+
+async def cmd_guard(ctx: CommandContext) -> OutboundMessage:
+    """Manage guardrails: enable/disable and banned topics."""
+    loop = ctx.loop
+    guardrail = getattr(loop, "_invariant_guardrail", None)
+    scanner = getattr(loop, "_llm_guard", None)
+
+    args = ctx.args.strip()
+
+    # /guard on — enable guardrails at runtime
+    if args == "on":
+        if guardrail is None or scanner is None:
+            # First-time enable — instantiate from config
+            from facio.config.loader import get_config_path, load_config, save_config
+            from facio.security.guardrails import InvariantGuardrail
+            from facio.security.llm_guard import LLMGuardScanner
+
+            config = load_config(get_config_path())
+            config.tools.guardrails.enabled = True
+            save_config(config, get_config_path())
+            guardrail = InvariantGuardrail(config.tools.guardrails)
+            scanner = LLMGuardScanner(config.tools.guardrails, getattr(loop, "workspace", None))
+            loop._invariant_guardrail = guardrail
+            loop._llm_guard = scanner
+        else:
+            guardrail._enabled = True
+            scanner._enabled = True
+            from facio.config.loader import get_config_path, load_config, save_config
+            config = load_config(get_config_path())
+            config.tools.guardrails.enabled = True
+            save_config(config, get_config_path())
+        return OutboundMessage(
+            channel=ctx.msg.channel,
+            chat_id=ctx.msg.chat_id,
+            content="✅ Guardrails **enabled**.",
+            metadata={"render_as": "text"},
+        )
+
+    # /guard off — disable guardrails at runtime
+    if args == "off":
+        if guardrail is not None:
+            guardrail._enabled = False
+        if scanner is not None:
+            scanner._enabled = False
+        from facio.config.loader import get_config_path, load_config, save_config
+        config = load_config(get_config_path())
+        config.tools.guardrails.enabled = False
+        save_config(config, get_config_path())
+        return OutboundMessage(
+            channel=ctx.msg.channel,
+            chat_id=ctx.msg.chat_id,
+            content="⛔ Guardrails **disabled**.",
+            metadata={"render_as": "text"},
+        )
+
+    # /guard status — show current state
+    if args == "status":
+        enabled = (guardrail is not None and guardrail.available) or (scanner is not None and scanner.available)
+        return OutboundMessage(
+            channel=ctx.msg.channel,
+            chat_id=ctx.msg.chat_id,
+            content=f"Guardrails: **{'enabled' if enabled else 'disabled'}**",
+            metadata={"render_as": "text"},
+        )
+
+    # Everything below requires guardrails to be active
+    if scanner is None or not scanner.available:
+        return OutboundMessage(
+            channel=ctx.msg.channel,
+            chat_id=ctx.msg.chat_id,
+            content="Guardrails are disabled. Use `/guard on` to enable.",
+            metadata={"render_as": "text"},
+        )
+
+    if not args or args == "list":
+        topics = scanner.ban_topics
+        if topics:
+            lines = "\n".join(f"- {t}" for t in topics)
+            content = f"**Banned topics ({len(topics)}):**\n{lines}"
+        else:
+            content = "No banned topics configured."
+        return OutboundMessage(
+            channel=ctx.msg.channel,
+            chat_id=ctx.msg.chat_id,
+            content=content,
+            metadata={"render_as": "text"},
+        )
+
+    parts = args.split(None, 1)
+    action = parts[0].lower()
+    topic = parts[1].strip() if len(parts) > 1 else ""
+
+    if action == "add":
+        if not topic:
+            return OutboundMessage(
+                channel=ctx.msg.channel,
+                chat_id=ctx.msg.chat_id,
+                content="Usage: `/guard add <topic>`",
+                metadata={"render_as": "text"},
+            )
+        scanner.add_topic(topic)
+        return OutboundMessage(
+            channel=ctx.msg.channel,
+            chat_id=ctx.msg.chat_id,
+            content=f"Added banned topic: **{topic}**",
+            metadata={"render_as": "text"},
+        )
+
+    if action == "remove":
+        if not topic:
+            return OutboundMessage(
+                channel=ctx.msg.channel,
+                chat_id=ctx.msg.chat_id,
+                content="Usage: `/guard remove <topic>`",
+                metadata={"render_as": "text"},
+            )
+        if topic.lower() in [t.lower() for t in scanner.ban_topics]:
+            scanner.remove_topic(topic)
+            return OutboundMessage(
+                channel=ctx.msg.channel,
+                chat_id=ctx.msg.chat_id,
+                content=f"Removed banned topic: **{topic}**",
+                metadata={"render_as": "text"},
+            )
+        return OutboundMessage(
+            channel=ctx.msg.channel,
+            chat_id=ctx.msg.chat_id,
+            content=f"Topic not found: **{topic}**",
+            metadata={"render_as": "text"},
+        )
+
+    return OutboundMessage(
+        channel=ctx.msg.channel,
+        chat_id=ctx.msg.chat_id,
+        content="Usage: `/guard [on|off|status|list|add|remove] [topic]`",
+        metadata={"render_as": "text"},
+    )

facio/command/builtins/guardrails.py

@@ -0,0 +1,70 @@
+"""Conversation history slash command handlers."""
+
+from __future__ import annotations
+
+from facio.bus.events import OutboundMessage
+from facio.command import responses
+from facio.command.router import CommandContext
+
+_HISTORY_DEFAULT_COUNT = 10
+_HISTORY_MAX_COUNT = 50
+_HISTORY_MAX_CONTENT_CHARS = 200
+
+
+def _format_history_message(msg: dict) -> str | None:
+    """Format a single history message for display. Returns None to skip."""
+    role = msg.get("role")
+    if role not in ("user", "assistant"):
+        return None
+    content = msg.get("content") or ""
+    if isinstance(content, list):
+        parts = [
+            b.get("text", "")
+            for b in content
+            if isinstance(b, dict) and b.get("type") == "text"
+        ]
+        content = " ".join(parts)
+    content = str(content).strip()
+    if not content:
+        return None
+    if len(content) > _HISTORY_MAX_CONTENT_CHARS:
+        content = content[:_HISTORY_MAX_CONTENT_CHARS] + "…"
+    label = "👤 You" if role == "user" else "🗿 facio"
+    return f"{label}: {content}"
+
+
+async def cmd_history(ctx: CommandContext) -> OutboundMessage:
+    """Show the last N messages of the current session (default 10, max 50).
+
+    Usage: /history [count]
+    """
+    count = _HISTORY_DEFAULT_COUNT
+    if ctx.args.strip():
+        try:
+            count = max(1, min(int(ctx.args.strip()), _HISTORY_MAX_COUNT))
+        except ValueError:
+            return OutboundMessage(
+                channel=ctx.msg.channel, chat_id=ctx.msg.chat_id,
+                content=responses.HISTORY_USAGE,
+                metadata=dict(ctx.msg.metadata or {}),
+            )
+
+    session = ctx.session or ctx.loop.sessions.get_or_create(ctx.key)
+    history = session.get_history(max_messages=_HISTORY_MAX_COUNT)
+    visible = [_format_history_message(m) for m in history]
+    visible = [m for m in visible if m is not None]
+    recent = visible[-count:]
+
+    if not recent:
+        return OutboundMessage(
+            channel=ctx.msg.channel, chat_id=ctx.msg.chat_id,
+            content=responses.NO_CONVERSATION_HISTORY,
+            metadata=dict(ctx.msg.metadata or {}),
+        )
+
+    header = responses.history_header(len(recent))
+    return OutboundMessage(
+        channel=ctx.msg.channel, chat_id=ctx.msg.chat_id,
+        content=header + "\n".join(recent),
+        metadata={**dict(ctx.msg.metadata or {}), "render_as": "text"},
+    )

facio/command/builtins/history.py

@@ -0,0 +1,112 @@
+"""Tool policy slash command handlers."""
+
+from __future__ import annotations
+
+import shlex
+
+from facio.bus.events import OutboundMessage
+from facio.command.router import CommandContext
+
+
+def _policy_text(ctx: CommandContext, content: str) -> OutboundMessage:
+    return OutboundMessage(
+        channel=ctx.msg.channel,
+        chat_id=ctx.msg.chat_id,
+        content=content,
+        metadata={"render_as": "text"},
+    )
+
+
+def _parse_policy_params(rest: str) -> dict[str, str]:
+    """Parse ``key=value`` tokens via :mod:`shlex` so values may be quoted.
+
+    Examples
+    --------
+    >>> _parse_policy_params('command="git status" cwd=/tmp')
+    {'command': 'git status', 'cwd': '/tmp'}
+    """
+    out: dict[str, str] = {}
+    if not rest.strip():
+        return out
+    try:
+        tokens = shlex.split(rest)
+    except ValueError:
+        # Unbalanced quotes — fall back to whitespace split so the user still
+        # gets a (best-effort) result instead of a hard crash.
+        tokens = rest.split()
+    for token in tokens:
+        if "=" not in token:
+            continue
+        k, _, v = token.partition("=")
+        k = k.strip()
+        if k:
+            out[k] = v.strip()
+    return out
+
+
+async def cmd_policy(ctx: CommandContext) -> OutboundMessage:
+    """Inspect and manage the tool-call policy (R11)."""
+    gate = getattr(ctx.loop, "policy_gate", None)
+    if gate is None:
+        return _policy_text(ctx, "Tool policy is disabled.")
+
+    args = ctx.args.strip()
+    parts = args.split(None, 1) if args else []
+    action = parts[0].lower() if parts else "list"
+    rest = parts[1] if len(parts) > 1 else ""
+
+    engine = gate.engine
+
+    if action in ("", "list", "status"):
+        rules = engine.rules
+        if not rules:
+            return _policy_text(ctx, "No policy rules configured.")
+        lines = [f"**Policy rules ({len(rules)}):**"]
+        for i, r in enumerate(rules, 1):
+            params = ", ".join(f"{k}={v}" for k, v in r.params.items()) or "*"
+            lines.append(f"{i}. `{r.action}` `{r.tool}` ({params})")
+        return _policy_text(ctx, "\n".join(lines))
+
+    if action == "clear":
+        engine.clear()
+        return _policy_text(ctx, "🗑️ Cleared all policy rules.")
+
+    if action in ("allow", "deny"):
+        sub = rest.split(None, 1)
+        if not sub:
+            return _policy_text(ctx, f"Usage: `/policy {action} <tool> [key=value ...]`")
+        tool = sub[0]
+        params = _parse_policy_params(sub[1] if len(sub) > 1 else "")
+        engine.add(action, tool, params, added_by="slash")
+        return _policy_text(ctx, f"✅ Added `{action}` rule for `{tool}`.")
+
+    if action == "remove":
+        sub = rest.split(None, 1)
+        if not sub:
+            return _policy_text(ctx, "Usage: `/policy remove <tool> [key=value ...]`")
+        tool = sub[0]
+        params = _parse_policy_params(sub[1] if len(sub) > 1 else "")
+        # Need an action — look up matching rule(s)
+        removed = 0
+        for r in list(engine.rules):
+            if r.tool == tool and r.params == params:
+                if engine.remove(r.action, r.tool, r.params):
+                    removed += 1
+        return _policy_text(ctx, f"Removed {removed} rule(s).")
+
+    return _policy_text(
+        ctx,
+        "Usage: `/policy [list|allow|deny|remove|clear] <tool> [key=value ...]`",
+    )
+
+
+async def cmd_allow(ctx: CommandContext) -> OutboundMessage:
+    """Shortcut: `/allow <tool> [key=value ...]`."""
+    ctx.args = f"allow {ctx.args}".strip()
+    return await cmd_policy(ctx)
+
+
+async def cmd_deny(ctx: CommandContext) -> OutboundMessage:
+    """Shortcut: `/deny <tool> [key=value ...]`."""
+    ctx.args = f"deny {ctx.args}".strip()
+    return await cmd_policy(ctx)