"""Unified server-side command parser.

Clients send {action: "command", text: "/compact grok", conversation_id: "..."}
and this module parses, dispatches, and returns the result.

This is the SINGLE source of truth for all /commands — webchat, vscode, and
pawcode CLI all use it. Client-only commands (clear-ui, connect, disconnect)
remain client-side.
"""

import json
import logging
import re
import shlex
from typing import Dict, Any, List, Optional, Callable

from core import FlowFile

logger = logging.getLogger(__name__)


# ── Help registry ─────────────────────────────────────────────────────

HELP: Dict[str, Dict[str, str]] = {
    # ── Conversation ──
    "/new": {
        "usage": "/new",
        "short": "Start a new conversation",
        "detail": "Creates a new conversation and switches to it.",
    },
    "/conv": {
        "usage": "/conv [list | select <id> | info]",
        "short": "List or switch conversations",
        "detail": (
            "  /conv list          — List all conversations\n"
            "  /conv select <id>   — Switch to a conversation\n"
            "  /conv info          — Show current conversation info"
        ),
        "aliases": "/conversations",
    },
    "/delete": {
        "usage": "/delete [conversation_id]",
        "short": "Delete a conversation",
        "detail": "Delete the current or specified conversation.",
    },
    "/rename": {
        "usage": "/rename <name>",
        "short": "Rename current conversation",
        "detail": "Set a display name for the current conversation.",
    },
    "/export": {
        "usage": "/export [json | md]",
        "short": "Export conversation",
        "detail": "Export the current conversation as JSON or Markdown.",
    },
    "/history": {
        "usage": "/history [count]",
        "short": "Show conversation messages",
        "detail": "Display the last N messages (default: 20).",
    },
    "/search": {
        "usage": "/search <text>",
        "short": "Search messages",
        "detail": "Search for text in all messages of the current conversation.",
    },
    "/fork": {
        "usage": "/fork [name]",
        "short": "Fork the current conversation",
        "detail": (
            "Create a copy of the current conversation with all messages and context.\n"
            "The original is preserved — you can explore a different direction.\n\n"
            "  /fork                — Fork with auto-generated name\n"
            "  /fork experiment-1   — Fork with a specific name"
        ),
        "aliases": "/branch",
    },

    # ── Agent management ──
    "/agent": {
        "usage": "/agent list | create | select | delete | msg | interrupt | btw | resume | setname",
        "short": "Manage AI agents",
        "detail": (
            "  /agent list                        — List all agents\n"
            "  /agent create                      — Create a new agent (interactive)\n"
            "  /agent select <name>               — Activate an agent\n"
            "  /agent delete <name>               — Delete an agent\n"
            "  /agent msg <name> <text>           — Message a specific agent\n"
            "  /agent msg ALL <text>              — Broadcast to all agents\n"
            "  /agent interrupt <name|ALL>        — Stop and respond immediately\n"
            "  /agent btw <name|ALL> <text>       — Side-channel question\n"
            "  /agent resume <name>               — Continue from where it stopped\n"
            "  /agent setname <real> [nickname]    — Set display nickname"
        ),
    },
    "/msg": {
        "usage": "/msg <name> <text>",
        "short": "Send message to specific agent",
        "detail": "Send a message without changing the active agent.\n\n  /msg grok Explain this code\n  /msg ALL What do you think?",
        "aliases": "/message",
    },
    "/btw": {
        "usage": "/btw <name|ALL> <question>",
        "short": "Side-channel question (no interruption)",
        "detail": "Quick question to an agent without interrupting its current work.",
    },
    "/stop": {
        "usage": "/stop [agent] [-f]",
        "short": "Stop agent execution",
        "detail": (
            "Stop the current or specified agent.\n\n"
            "  /stop            — Ask current agent to wrap up\n"
            "  /stop grok       — Ask specific agent to wrap up\n"
            "  /stop -f         — Force stop immediately (no response)\n"
            "  /stop grok -f    — Force stop specific agent"
        ),
    },
    "/resume": {
        "usage": "/resume [agent]",
        "short": "Tell agent to continue",
        "detail": "Tell the current or specified agent to continue from where it stopped.",
    },
    "/setname": {
        "usage": "/setname <agent> [nickname]",
        "short": "Set agent display nickname",
        "detail": "Set or reset (omit nickname) the display name for an agent.",
    },

    # ── Context management ──
    "/compact": {
        "usage": "/compact [agent] [instructions]",
        "short": "Compact context (summarize old messages)",
        "detail": (
            "Compress conversation context by summarizing old messages.\n\n"
            "  /compact              — Compact current agent's context\n"
            "  /compact grok         — Compact specific agent\n"
            "  /compact -- focus on the auth refactor  — Focus compaction on a topic\n\n"
            "Like Claude Code, you can add instructions to focus the compaction."
        ),
    },
    "/git-prune": {
        "usage": "/git-prune",
        "short": "Prune conversation Git snapshot history",
        "detail": (
            "Run conversation Git retention now for the current conversation.\n"
            "This blocks conversation context operations like /compact, rewrites "
            "the bounded live history, expires reflogs, and runs git gc to reclaim space."
        ),
        "aliases": "/prune-git",
    },
    "/context": {
        "usage": "/context [agent]",
        "short": "View the LLM context",
        "detail": "Show the current agent context (messages, tokens, etc.).",
    },
    "/model": {
        "usage": "/model <name> | reset",
        "short": "Switch LLM model",
        "detail": (
            "Override the model for the current agent in this conversation.\n\n"
            "  /model gpt-4o       — Use gpt-4o\n"
            "  /model reset        — Revert to default\n\n"
            "This changes the model, not the service. Use /llm to change the service."
        ),
    },
    "/llm": {
        "usage": "/llm <service_name> [agent]",
        "short": "Change LLM service for agent",
        "detail": (
            "  /llm grok_service        — Set for current agent\n"
            "  /llm grok_service claude  — Set for specific agent\n\n"
            "Supports expressions: /llm ${preferred_llm}"
        ),
        "aliases": "/set_llm_service",
    },
    "/effort": {
        "usage": "/effort low | medium | high | max | reset",
        "short": "Set thinking effort level",
        "detail": (
            "Control how much the model thinks before responding.\n\n"
            "  /effort low    — No extended thinking (budget=0)\n"
            "  /effort medium — Moderate thinking (budget=5000)\n"
            "  /effort high   — Deep thinking (budget=10000)\n"
            "  /effort max    — Maximum thinking (budget=20000)\n"
            "  /effort reset  — Revert to default\n\n"
            "Supports expressions: /effort ${default_effort}"
        ),
    },
    "/fast": {
        "usage": "/fast [on | off | <model>]",
        "short": "Toggle fast/cheap model",
        "detail": (
            "Switch to a fast/cheap model for quick tasks.\n\n"
            "  /fast           — Toggle on/off\n"
            "  /fast on        — Enable (uses configured fast model)\n"
            "  /fast off       — Disable (revert to normal model)\n"
            "  /fast gpt-4o-mini  — Use a specific fast model\n\n"
            "Configure default: /add-variable fast_model gpt-4o-mini"
        ),
    },
    "/rebuild": {
        "usage": "/rebuild",
        "short": "Rebuild shared context, buckets, and compacted agent contexts",
        "detail": "Rebuild shared context from transcript, rebuild buckets, then compact every conversation agent.",
    },
    "/rebuild-full": {
        "usage": "/rebuild-full [agent]",
        "short": "Set context = full conversation",
        "detail": "Set the agent context to the full conversation (no compaction).",
    },
    "/restart": {
        "usage": "/restart <index|msg_id>",
        "short": "Restart conversation from a point",
        "detail": (
            "Truncate transcript/shared context at an absolute index or msg_id.\n\n"
            "  /restart 0          — Empty transcript and contexts\n"
            "  /restart 10         — Keep the first 10 messages\n"
            "  /restart abc123     — Keep messages through msg_id abc123"
        ),
        "aliases": "/restart_from",
    },
    "/rewind": {
        "usage": "/rewind [checkpoint]",
        "short": "Rewind to a previous checkpoint",
        "detail": (
            "Undo file changes and/or conversation to a previous point.\n\n"
            "  /rewind         — Show available checkpoints\n"
            "  /rewind 3       — Rewind to checkpoint #3\n\n"
            "Options: restore code+conversation, conversation only, code only, or summarize."
        ),
        "aliases": "/checkpoint",
    },
    "/summary": {
        "usage": "/summary [tokens] [agent]",
        "short": "Summarize context to N tokens",
        "detail": "Summarize the context to approximately N tokens.",
    },
    "/cc_restart": {
        "usage": "/cc_restart [agent]",
        "short": "Kill warm Claude Code session(s) for this conv",
        "detail": (
            "Force a fresh Claude Code subprocess spawn on the next turn.\n\n"
            "  /cc_restart          — kill every live CC session in this conv\n"
            "  /cc_restart claude   — kill only the 'claude' agent's session"
        ),
    },
    "/cc_live": {
        "usage": "/cc_live [agent]",
        "short": "Show warm Claude Code sessions for this conv",
        "detail": (
            "List live CC subprocesses still attached to this conv:\n"
            "idle time, reuse count, spawn age, service/pool index."
        ),
    },
    "/codex_restart": {
        "usage": "/codex_restart [agent]",
        "short": "Kill warm codex container(s) for this conv",
        "detail": (
            "Force a fresh codex container spawn on the next turn.\n\n"
            "  /codex_restart        — kill every live codex container in this conv\n"
            "  /codex_restart agent  — kill only that agent's container"
        ),
    },
    "/codex_live": {
        "usage": "/codex_live [agent]",
        "short": "Show warm codex containers for this conv",
        "detail": (
            "List live codex containers still pinned to this conv:\n"
            "idle time, reuse count, spawn age, service."
        ),
    },
    "/gemini_restart": {
        "usage": "/gemini_restart [agent]",
        "short": "Kill warm gemini container(s) for this conv",
        "detail": (
            "Force a fresh gemini container spawn on the next turn.\n\n"
            "  /gemini_restart        — kill every live gemini container in this conv\n"
            "  /gemini_restart agent  — kill only that agent's container"
        ),
    },
    "/gemini_live": {
        "usage": "/gemini_live [agent]",
        "short": "Show warm gemini containers for this conv",
        "detail": (
            "List live gemini containers still pinned to this conv:\n"
            "idle time, reuse count, spawn age, service."
        ),
    },

    # ── Resources ──
    "/resources": {
        "usage": "/resources",
        "short": "List all resources",
        "detail": "Show all defined resources (agents, skills, MCP servers) with status.",
    },
    "/tools": {
        "usage": "/tools",
        "short": "List available tools",
        "detail": "List all tools available to the current agent.",
    },
    "/tool-metrics": {
        "usage": "/tool-metrics",
        "short": "Show tool execution metrics",
        "detail": "Show per-tool call counts, errors, latency, and the latest error.",
    },
    "/toolmetrics": {
        "usage": "/toolmetrics",
        "short": "Show tool execution metrics",
        "detail": "Alias for /tool-metrics.",
    },
    "/call": {
        "usage": '/call tool_name(key=value, ...) or /call tool_name {"key": "value"}',
        "short": "Call a tool directly",
        "detail": (
            "Execute any agent tool from the chat.\n\n"
            "  /call web_search(query=\"quantum computing\")\n"
            "  /call list_dir(path=\"/home\")\n"
            "  /call web_search {\"query\": \"quantum computing\"}"
        ),
    },
    "/skill": {
        "usage": "/skill list | search [--source src] <query> | import [--source src] [--review-only] [--force] [--scope global|user|conversation] [--name name] <ref> | add [--force] <name> <prompt> | update [--force] <name> <prompt> | del <name> | assign @agent @skill | unassign @agent @skill | assigned @agent | run [@agent] <name> [args...] | //<name> [@agent] [args...]",
        "short": "Manage skills",
        "detail": (
            "  /skill list                    — List all skills\n"
            "  /skill search [--source src] <query> — Search external skill marketplaces\n"
            "  /skill import [--source src] [--review-only] [--force] [--scope global|user|conversation] [--name name] <ref> — Review/import an external skill\n"
            "  /skill add [--force] <name> <prompt> — Create a skill\n"
            "  /skill update [--force] <name> <prompt> — Update a skill\n"
            "  /skill del <name>              — Delete a skill\n"
            "  /skill assign @agent @skill    — Assign a skill to an agent\n"
            "  /skill unassign @agent @skill  — Remove a skill from an agent\n"
            "  /skill assigned @agent         — List skills assigned to an agent\n"
            "  /skill run [@agent] <name> [args...] — Invoke a skill now\n"
            "  //<name> [@agent] [args...]    — Shortcut for /skill run"
        ),
    },
    "/pfp": {
        "usage": "/pfp inspect|install|update|build|dev-load|dev-unload|export|uninstall|list|reload-tasks|search|registry|key-create ...",
        "short": "Manage PawFlow packages",
        "detail": (
            "  /pfp key-create                         — Create an Ed25519 signing key\n"
            "  /pfp build <pfpdir> --key-env VAR [--out file.pfp]\n"
            "  /pfp inspect <file.pfp|pfpdir|ref|url> [--confirm-download] — Verify and preview objects/capabilities\n"
            "  /pfp install <file.pfp|ref|url> [--confirm-download] [--scope user|conversation] [--include ids] [--secret logical=stored_key] [--force]\n"
            "  /pfp dev-load <pfpdir> [--scope conversation|user] [--include ids] [--exclude ids] [--secret logical=stored_key] [--replace]\n"
            "  /pfp dev-unload <package> [--scope conversation|user]\n"
            "  /pfp update <file.pfp|ref|url> [--confirm-download] [--include ids] [--exclude ids] [--force]\n"
            "  /pfp search <query>                     — Search configured decentralized registries\n"
            "  /pfp registry add <url> [--name name] [--trusted] — Add a static registry index\n"
            "  /pfp registry list|remove <name-or-url> — Manage configured registries\n"
            "  /pfp uninstall <package> [--scope user|conversation]\n"
            "  /pfp list [--scope user|conversation]   — List installed packages\n"
            "  /pfp reload-tasks [--scope user|conversation] — Reload installed package task proxies\n"
            "  /pfp export --package id --version v --include type:name[,type:name] --out dir"
        ),
    },
    "/task": {
        "usage": "/task create | assign | list | delete | pause | resume | cancel",
        "short": "Manage agent tasks",
        "detail": (
            "  /task create <name> \"<prompt>\" [--criteria \"...\"] [--interval XX]\n"
            "  /task assign <agent> <name> [--var k=v] [--verifier <agent>]\n"
            "  /task list            — Show library + running tasks\n"
            "  /task pause <id>      — Pause a task\n"
            "  /task resume <id>     — Resume a paused task\n"
            "  /task cancel <id>     — Cancel a task"
        ),
    },
    "/goal": {
        "usage": "/goal [@agent] \"<objective>\" [task options]",
        "short": "Create and assign a conversation-scoped goal task",
        "detail": (
            "Create a conversation-scoped task definition with a generated name "
            "and assign it immediately. If no agent is provided, the selected "
            "conversation agent is used. The objective is also used as criteria "
            "unless --criteria is provided.\n\n"
            "Options mirror /task assign: --criteria, --interval, --verifier, "
            "--budget, --turn-time, --total-time, --max-reschedules, --max, "
            "--context, --var, --auto-allow, --interactive."
        ),
    },
    "/permission": {
        "usage": "/permission [default | approve_edits | read_only | auto | tool <name> allow|deny|confirm|reset | tools]",
        "short": "Set tool permission mode or per-tool permissions",
        "detail": (
            "Set how tool calls are authorized for this conversation:\n\n"
            "  /permission                              — Show current mode\n"
            "  /permission default                      — Normal approval gate\n"
            "  /permission approve_edits                — Same as default (explicit approval)\n"
            "  /permission read_only                    — Block all write operations\n"
            "  /permission auto                         — Auto-approve everything (no prompts)\n"
            "  /permission tool <name> allow            — Always allow this tool (no prompt)\n"
            "  /permission tool <name> deny             — Always block this tool\n"
            "  /permission tool <name> confirm          — Always ask for confirmation\n"
            "  /permission tool <name> reset            — Remove per-tool override\n"
            "  /permission tools                        — List per-tool overrides"
        ),
    },
    "/service": {
        "usage": "/service list | add | delete | test",
        "short": "Manage LLM and external services",
        "detail": "Manage LLM services, image/video services, filesystem services, etc.",
    },
    "/workspace": {
        "usage": "/workspace [status | destroy]",
        "short": "Server-side workspace (Docker, persistent)",
        "detail": (
            "  /workspace          — Create a server workspace for this conversation\n"
            "  /workspace status   — Show workspace status\n"
            "  /workspace destroy  — Stop container and delete workspace volume"
        ),
    },
    "/relay": {
        "usage": "/relay [list | link <id> | unlink <id> | default <id>]",
        "short": "Manage relay bindings for this conversation",
        "detail": (
            "  /relay              — List relays linked to this conversation\n"
            "  /relay list         — List all available relays\n"
            "  /relay link <id>    — Link a relay to this conversation\n"
            "  /relay unlink <id>  — Unlink a relay from this conversation\n"
            "  /relay default <id> — Set the default relay for this conversation"
        ),
    },
    "/flow": {
        "usage": "/flow list | templates | deploy | start | stop | undeploy | promote",
        "short": "Manage data flows",
        "detail": "Manage NiFi-style data processing flows.",
    },
    "/prompt": {
        "usage": "/prompt list | use <name>",
        "short": "Manage prompts",
        "detail": "List available prompts or inject one into the conversation.",
    },
    "/memory": {
        "usage": "/memory list | add | edit | del | search",
        "short": "Manage agent memories",
        "detail": (
            "  /memory list [agent]      — List memories\n"
            "  /memory add <text>        — Add a memory\n"
            "  /memory search <query>    — Search memories\n"
            "  /memory del <id>          — Delete a memory"
        ),
    },
    "/cost": {
        "usage": "/cost [agent]",
        "short": "Show token usage and estimated cost",
        "detail": (
            "Show token usage per agent and estimated cost for the session.\n"
            "If max_budget_usd is configured on the LLM service, also shows remaining budget.\n"
            "A budget_warning SSE event fires at 80% usage. Agent stops at 100%."
        ),
        "aliases": "/usage",
    },

    "/image": {
        "usage": "/image [@service] prompt [--width N] [--height N] [--style S] [--negative_prompt S]",
        "short": "Generate an image",
        "detail": "Generate an image using the configured image service. Use @service to pick a specific service.",
    },
    "/video": {
        "usage": "/video [@service] prompt [--duration N] [--width N] [--height N]",
        "short": "Generate a video",
        "detail": "Generate a video using the configured video service. Use @service to pick a specific service.",
    },
    "/connect": {
        "usage": "/connect [@relay_source] /path/to/dir",
        "short": "Add a filesystem service from a connected relay",
        "detail": (
            "Spawns a new relay process on the remote machine for the given directory.\n"
            "Use @relay to specify which connected relay to use. If omitted, uses the user's default relay."
        ),
    },
    "/disconnect": {
        "usage": "/disconnect @service",
        "short": "Disconnect and remove a filesystem service",
        "detail": "Stops the remote relay process and removes the service.",
    },
    "/audio": {
        "usage": "/audio [@service] prompt [--duration N] [--style S] [--instrumental] [--lyrics TEXT]",
        "short": "Generate audio/music",
        "detail": "Generate audio or music using the configured audio service. Use @service to pick a specific service.",
    },
    "/claude-code-auth": {
        "usage": "/claude-code-auth @service_name {credentials JSON}",
        "short": "Authenticate Claude Code with subscription credentials",
        "detail": (
            "Paste the content of ~/.claude/.credentials.json after running `claude auth login` locally.\n"
            "Admin required for global services. Users can auth their own services."
        ),
    },

    # ── Secrets & Variables ──
    "/secrets": {
        "usage": "/secrets",
        "short": "List stored secrets",
        "detail": "List all encrypted secrets (values are hidden).",
        "aliases": "/list-secrets",
    },
    "/add-secret": {
        "usage": "/add-secret <name> <value>",
        "short": "Store an encrypted secret",
        "detail": "Store a secret value (API key, password, etc.). Encrypted at rest.",
    },
    "/variables": {
        "usage": "/variables",
        "short": "List stored variables",
        "detail": "List all plaintext variables.",
        "aliases": "/vars, /list-variables",
    },
    "/add-variable": {
        "usage": "/add-variable <name> <value>",
        "short": "Store a plaintext variable",
        "detail": "Store a variable accessible via ${name} in expressions.",
        "aliases": "/add-var",
    },

    # ── Scheduling ──
     "/schedules": {
         "usage": "/schedules list | add <datetime> [reason] [@agent] [--loop <seconds>] | del <key>",
         "short": "Manage scheduled wake-ups",
         "detail": "Manage scheduled poll rechecks for agents.\n\n"
                   "  /schedules list                     — List pending schedules\n"
                   "  /schedules add <YYYYMMDDHHmmss> [reason] [@agent] [--loop N]\n"
                   "                                      — Add a schedule (optionally recurring every N seconds)\n"
                   "  /schedules del <key>                — Delete a specific schedule by key\n"
                   "  /schedules del all                  — Delete all schedules",
    },
    "/autoconv": {
        "usage": "/autoconv <agent> [on|off|status] [min] [max]",
        "short": "Auto-conversation (agent thinks autonomously)",
        "detail": (
            "Enable agents to contribute thoughts at random intervals.\n\n"
            "  /autoconv grok on 30 120   — grok thinks every 30-120s\n"
            "  /autoconv grok off         — Disable\n"
            "  /autoconv grok status      — Show current settings"
        ),
    },
    "/loop": {
        "usage": "/loop [interval] <prompt or /command>",
        "short": "Run a prompt on a recurring interval",
        "detail": (
            "  /loop 5m check if deploy finished\n"
            "  /loop 10m /cost\n\n"
            "Default interval: 10 minutes."
        ),
    },

    # ── Files ──
    "/files": {
        "usage": "/files",
        "short": "Toggle the files panel",
        "detail": "Show/hide the files panel listing conversation attachments.",
    },
    "/upload": {
        "usage": "/upload",
        "short": "Upload a file",
        "detail": "Open file picker to upload an attachment.",
    },
    "/paste": {
        "usage": "/paste",
        "short": "Paste clipboard content",
        "detail": "Paste clipboard text as a message or attachment.",
    },
    "/copy": {
        "usage": "/copy",
        "short": "Copy last response to clipboard",
        "detail": "Copy the last assistant response to the clipboard.",
    },
    "/view": {
        "usage": "/view <file_id or path>",
        "short": "Preview a file",
        "detail": "Preview an image, PDF, text, or code file.",
    },
    "/run": {
        "usage": "/run [@service] <command>",
        "short": "Execute shell command via relay",
        "detail": (
            "Run a shell command through the connected relay.\n"
            "Output is shown directly, not sent to the agent.\n\n"
            "  /run git status\n"
            "  /run @my-relay ls -la\n"
            "  /run npm test\n\n"
            "If multiple relays are connected, use @service to pick one."
        ),
    },
    "/diff": {
        "usage": "/diff",
        "short": "Show git diff",
        "detail": "Show uncommitted changes in the connected workspace.",
    },
    "/plan": {
        "usage": "/plan [on | off | status]",
        "short": "Plan mode (approval before execution)",
        "detail": (
            "Toggle plan mode — agent proposes actions and waits for approval.\n\n"
            "  /plan          — Toggle on/off\n"
            "  /plan on       — Enable plan mode\n"
            "  /plan off      — Disable plan mode\n"
            "  /plan status   — Show current state"
        ),
    },

    # ── Activation & Sharing ──
    "/activate": {
        "usage": "/activate <type> <name>",
        "short": "Activate a resource",
        "detail": "Activate an agent, skill, or MCP server for this conversation.",
    },
    "/deactivate": {
        "usage": "/deactivate <type> <name>",
        "short": "Deactivate a resource",
        "detail": "Deactivate a resource from this conversation.",
    },
    "/share": {
        "usage": "/share <type> <name> <target_conv>",
        "short": "Share a resource",
        "detail": "Share a resource to another conversation.",
    },
    "/link": {
        "usage": "/link <provider> <id>",
        "short": "Link/unlink external accounts",
        "detail": "Link an external account (Telegram, Discord, etc.) to your PawFlow user.",
    },

    # ── Session ──
    "/login": {
        "usage": "/login",
        "short": "Re-authenticate",
        "detail": "Re-authenticate your session.",
    },
    "/help": {
        "usage": "/help [command]",
        "short": "Show available commands",
        "detail": "Show all commands or detailed help for a specific command.",
    },
    "/doctor": {
        "usage": "/doctor",
        "short": "Diagnose system health",
        "detail": (
            "Check the health of all system components:\n"
            "  — LLM services (configured, reachable)\n"
            "  — Relay connection\n"
            "  — FileStore\n"
            "  — Agent definitions\n"
            "  — Authentication"
        ),
    },
    "/add-dir": {
        "usage": "/add-dir <path>",
        "short": "Add a workspace directory",
        "detail": (
            "Create a filesystem service pointing to the given path.\n\n"
            "  /add-dir /home/user/project\n"
            "  /add-dir .                    — Current relay directory"
        ),
    },

    # ── Stats & Analysis ──
    "/stats": {
        "usage": "/stats",
        "short": "Show session statistics",
        "detail": "Token usage per agent, model breakdown, message counts.",
    },
    "/pr-comments": {
        "usage": "/pr-comments [PR number or URL]",
        "short": "Fetch GitHub PR comments",
        "detail": "Fetch and display comments from a GitHub pull request.",
    },
    "/security-review": {
        "usage": "/security-review",
        "short": "Security audit of pending changes",
        "detail": "Analyze git diff for security vulnerabilities (injection, auth, secrets, etc.).",
    },
    "/insights": {
        "usage": "/insights",
        "short": "Generate session insights",
        "detail": "Analyze conversation history for patterns, friction points, and suggestions.",
    },
    "/feedback": {
        "usage": "/feedback <description>",
        "short": "Report an issue or give feedback",
        "detail": "Report a bug or suggest an improvement.",
        "aliases": "/bug",
    },

    # ── Developer ──
    "/install": {
        "usage": "/install <tool_url_or_path>",
        "short": "Install a custom tool",
        "detail": "Install a custom tool handler from a URL or file path.",
    },
    "/uninstall": {
        "usage": "/uninstall <tool_name>",
        "short": "Uninstall a custom tool",
        "detail": "Remove a custom tool handler.",
    },
    "/batch": {
        "usage": "/batch <instruction>",
        "short": "Parallel changes across multiple files",
        "detail": "Instruct the agent to make coordinated changes across multiple files.",
    },
    "/debug": {
        "usage": "/debug",
        "short": "Diagnose session issues",
        "detail": "Run diagnostics on the current session (context, services, etc.).",
    },
    "/clear-store": {
        "usage": "/clear-store [agent | ALL]",
        "short": "Clean up FileStore files",
        "detail": "Delete stored files for the current agent or all agents.",
    },
    "/hooks": {
        "usage": "/hooks [list | add | del]",
        "short": "Manage tool execution hooks",
        "detail": (
            "Configure scripts that run before/after tool execution.\n\n"
            "  /hooks list                           — List active hooks\n"
            "  /hooks add pre:write eslint --fix ${path}\n"
            "  /hooks del <hook_id>"
        ),
    },
}

# Build alias map
ALIASES: Dict[str, str] = {}
for _cmd, _info in HELP.items():
    for _alias in (_info.get("aliases") or "").split(","):
        _alias = _alias.strip()
        if _alias:
            ALIASES[_alias] = _cmd


# ── Effort mapping ────────────────────────────────────────────────────

EFFORT_MAP = {
    "low": "0",
    "medium": "5000",
    "high": "10000",
    "max": "20000",
}


# ── Command parser ────────────────────────────────────────────────────

import re as _re_cmd

def _extract_at_agent(arg: str, default_agent: str) -> tuple:
    """Extract @agent from command arguments.

    Returns (agent_name, remaining_arg).
    @agent can appear anywhere in the arg. If not present, uses default_agent.
    "ALL" is a special target (broadcast).
    """
    m = _re_cmd.search(r'@(\S+)', arg)
    if m:
        agent = m.group(1)
        remaining = (arg[:m.start()] + arg[m.end():]).strip()
        return agent, remaining
    return default_agent, arg


def _parse_command(text: str, conversation_id: str, user_id: str,
                   agent_name: str = "") -> Optional[Dict[str, Any]]:
    """Parse a /command text into an action body dict.

    Returns None if the command is not recognized or is client-only.
    Returns {"_client_only": True, ...} for commands handled client-side.

    Convention: @agent in any argument targets that agent.
    If no @agent, uses the currently selected agent.
    """
    text = text.strip()
    if not text.startswith("/"):
        return None

    # Skill-run sugar: //skill-name [@agent] [args...]
    if text.startswith("//"):
        return _parse_skill_sugar_command(
            text, {"conversation_id": conversation_id}, agent_name)

    # Split command and args
    parts = text.split(None, 1)
    cmd = parts[0].lower()
    arg = parts[1] if len(parts) > 1 else ""

    # Resolve aliases
    cmd = ALIASES.get(cmd, cmd)

    # Common fields
    base = {"conversation_id": conversation_id}

    # ── Conversation ──
    if cmd == "/new":
        return {"action": "new_conversation", **base}

    if cmd == "/delete":
        target = arg.strip() or conversation_id
        return {"action": "delete_conversation", "conversation_id": target}

    if cmd == "/rename":
        return {"action": "rename_conversation", "name": arg.strip(), **base}

    if cmd == "/export":
        fmt = arg.strip().lower() or "markdown"
        return {"action": "export", "format": fmt, **base}

    if cmd == "/history":
        limit = int(arg.strip()) if arg.strip().isdigit() else 20
        return {"action": "load_history", "limit": limit, "offset": 0, **base}

    if cmd == "/search":
        return {"action": "search_messages", "query": arg.strip(), **base}

    if cmd == "/fork":
        return {"action": "fork_conversation", "name": arg.strip(), **base}

    # ── Context ──
    if cmd == "/compact":
        # Parse: /compact [@agent] [--force] [-- instructions]
        agent, rest = _extract_at_agent(arg, agent_name)
        force = False
        if "--force" in rest:
            force = True
            rest = rest.replace("--force", "").strip()
        instructions = ""
        if "--" in rest:
            _, instructions = rest.split("--", 1)
            instructions = instructions.strip()
        return {"action": "compact", "agent_name": agent,
                "instructions": instructions, "force": force, **base}

    if cmd == "/git-prune":
        return {"action": "git_prune", **base}

    if cmd == "/context":
        agent, _ = _extract_at_agent(arg, agent_name)
        return {"action": "view_context", "agent_name": agent, **base}

    if cmd == "/model":
        return {"action": "model", "model": arg.strip(), "agent": agent_name,
                **base}

    if cmd in ("/llm", "/set_llm_service"):
        agent, svc = _extract_at_agent(arg, agent_name)
        return {"action": "set_llm_service", "service": svc.strip(), "agent": agent,
                **base}

    if cmd == "/effort":
        val = arg.strip().lower()
        if val == "reset":
            return {"action": "set_effort", "value": "reset", **base}
        resolved = EFFORT_MAP.get(val, val)
        return {"action": "set_effort", "value": resolved, **base}

    if cmd == "/fast":
        val = arg.strip().lower()
        if not val or val == "on":
            return {"action": "set_fast", "enabled": True, **base}
        if val == "off":
            return {"action": "set_fast", "enabled": False, **base}
        # Specific model name
        return {"action": "set_fast", "enabled": True, "model": val, **base}

    if cmd == "/rebuild":
        return {"action": "rebuild", **base}

    if cmd == "/rebuild-full":
        agt, _ = _extract_at_agent(arg, agent_name)
        return {"action": "rebuild_full", "agent_name": agt, **base}

    if cmd in ("/restart", "/restart_from"):
        p = arg.split()
        restart_index = None
        restart_msg_id = ""
        for part in p:
            if part.startswith("@"):
                continue
            if part.isdigit():
                restart_index = int(part)
            else:
                restart_msg_id = part
        out = {"action": "restart_from", **base}
        if restart_msg_id:
            out["msg_id"] = restart_msg_id
        elif restart_index is not None:
            out["restart_index"] = restart_index
        else:
            out["error"] = "Usage: /restart_from <index|msg_id>"
        return out

    if cmd in ("/rewind", "/checkpoint"):
        return {"action": "rewind", "checkpoint": arg.strip(), **base}

    if cmd == "/summary":
        agt, rest = _extract_at_agent(arg, agent_name)
        tokens = 500
        for part in rest.split():
            if part.isdigit():
                tokens = int(part)
        return {"action": "resume_conversation", "max_tokens": tokens,
                "agent_name": agt, **base}

    if cmd == "/cc_restart":
        # Optional single positional arg = agent name. Empty = all agents.
        _arg = arg.strip()
        agt = _arg if _arg else ""
        return {"action": "cc_restart", "agent_name": agt, **base}

    if cmd == "/cc_live":
        _arg = arg.strip()
        agt = _arg if _arg else ""
        return {"action": "cc_live_status", "agent_name": agt, **base}

    if cmd == "/codex_restart":
        _arg = arg.strip()
        agt = _arg if _arg else ""
        return {"action": "codex_restart", "agent_name": agt, **base}

    if cmd == "/codex_live":
        _arg = arg.strip()
        agt = _arg if _arg else ""
        return {"action": "codex_live_status", "agent_name": agt, **base}

    if cmd == "/gemini_restart":
        _arg = arg.strip()
        agt = _arg if _arg else ""
        return {"action": "gemini_restart", "agent_name": agt, **base}

    if cmd == "/gemini_live":
        _arg = arg.strip()
        agt = _arg if _arg else ""
        return {"action": "gemini_live_status", "agent_name": agt, **base}

    # ── Agent ──
    if cmd == "/agent":
        return _parse_agent_command(arg, base, agent_name)

    if cmd in ("/msg", "/message"):
        target, message = _extract_at_agent(arg, agent_name)
        if target.upper() == "ALL":
            return {"action": "broadcast_agents", "message": message, **base}
        return {"action": "agent_msg", "target_agent": target,
                "message": message, **base}

    if cmd == "/btw":
        target, question = _extract_at_agent(arg, agent_name)
        return {"action": "btw", "agent_name": target, "question": question,
                **base}

    if cmd == "/stop":
        _stop_arg = arg.replace("-f", "").strip()
        agt, _ = _extract_at_agent(_stop_arg, agent_name)
        return {"action": "cancel", "agent_name": agt, **base}

    if cmd == "/resume":
        agt, _ = _extract_at_agent(arg, agent_name)
        return {"action": "resume_agent", "agent_name": agt, **base}

    if cmd == "/setname":
        real, nick = _extract_at_agent(arg, agent_name)
        return {"action": "set_nickname", "agent_name": real, "nickname": nick.strip(),
                **base}

    # ── Filesystem relay ──
    if cmd == "/connect":
        # /connect @relay_source /path/to/dir — or just /connect /path (uses default relay)
        svc, path = _extract_at_agent(arg, "")
        return {"action": "relay_connect", "relay_source": svc,
                "path": path.strip(), **base}

    if cmd == "/disconnect":
        svc, _ = _extract_at_agent(arg, "")
        return {"action": "relay_disconnect", "service_id": svc, **base}

    # ── Resources ──
    if cmd == "/resources":
        return {"action": "list_resources", **base}

    if cmd == "/tools":
        return {"action": "list_tools", **base}

    if cmd in ("/tool-metrics", "/toolmetrics"):
        return {"action": "tool_metrics", **base}

    if cmd == "/call":
        return {"action": "user_tool_call", "call_text": arg, **base}

    if cmd == "/skill":
        return _parse_skill_command(arg, base, agent_name)

    if cmd == "/add-skill":
        return _parse_skill_command("add " + arg, base, agent_name)

    if cmd == "/pfp":
        return _parse_pfp_command(arg, base)

    if cmd == "/task":
        return _parse_task_command(arg, base)

    if cmd == "/goal":
        return _parse_goal_command(arg, base, agent_name)

    if cmd == "/permission":
        parts = arg.strip().split() if arg.strip() else []
        if not parts:
            return {"action": "get_permission_mode", **base}
        subcmd = parts[0].lower()
        if subcmd == "tools":
            return {"action": "get_tool_permissions", **base}
        if subcmd == "tool":
            if len(parts) < 3:
                return {"display": "Usage: /permission tool <name> allow|deny|confirm|reset"}
            tool_name = parts[1]
            perm = parts[2].lower()
            if perm == "reset":
                perm = ""
            elif perm not in ("allow", "deny", "confirm"):
                return {"display": f"Invalid permission: {perm}. Use allow|deny|confirm|reset"}
            return {"action": "set_tool_permission", "tool_name": tool_name, "permission": perm, **base}
        # Global mode
        mode = subcmd
        if mode not in ("default", "approve_edits", "read_only", "auto"):
            return {"display": f"Invalid mode: {mode}. Valid: default, approve_edits, read_only, auto"}
        return {"action": "set_permission_mode", "mode": mode, **base}

    if cmd == "/service":
        return _parse_service_command(arg, base, user_id)

    if cmd == "/flow":
        return _parse_flow_command(arg, base)

    if cmd == "/prompt":
        p = arg.split(None, 1)
        subcmd = p[0] if p else "list"
        if subcmd == "use":
            return {"action": "use_prompt", "name": p[1] if len(p) > 1 else "",
                    **base}
        return {"action": "list_prompts", **base}

    if cmd in ("/image", "/video", "/audio"):
        # /image [@service] prompt text [--param value ...]
        # /video [@service] prompt text [--param value ...]
        # /audio [@service] prompt text [--param value ...]
        _media_map = {"/image": "generate_image", "/video": "generate_video", "/audio": "generate_audio"}
        tool_name = _media_map[cmd]
        svc, rest = _extract_at_agent(arg, "")  # @service, not @agent
        # Parse --key value params from the rest (supports multiline prompts)
        params = {}
        # Split on " --" or "\n--" to separate prompt from params
        import re as _re_media
        parts = _re_media.split(r'(?:^|\s)--', rest, maxsplit=0)
        params["prompt"] = parts[0].strip()
        for part in parts[1:]:
            kv = part.split(None, 1)
            if len(kv) == 2:
                params[kv[0]] = kv[1].strip().strip('"').strip("'")
            elif kv:
                params[kv[0]] = "true"
        # Convert numeric params
        for _nk in ("width", "height", "num_inference_steps", "duration"):
            if _nk in params:
                try:
                    params[_nk] = int(params[_nk])
                except ValueError:
                    pass
        for _fk in ("guidance_scale",):
            if _fk in params:
                try:
                    params[_fk] = float(params[_fk])
                except ValueError:
                    pass
        # Convert boolean params
        for _bk in ("instrumental",):
            if _bk in params:
                params[_bk] = params[_bk].lower() in ("true", "1", "yes", "")
        if svc:
            params["_service"] = svc
        # Dispatch as call_tool — same path as /call
        return {"action": "call_tool", "tool_name": tool_name,
                "arguments": params, **base}

    if cmd == "/claude-code-auth":
        # /claude-code-auth @service_name {credentials JSON}
        # Extract service name via @, rest is the credentials JSON
        svc_name, creds = _extract_at_agent(arg, "")
        return {"action": "claude_code_auth", "service_id": svc_name,
                "credentials": creds.strip(), **base}

    if cmd == "/memory":
        return _parse_memory_command(arg, base, agent_name)

    if cmd in ("/cost", "/usage"):
        agt, _ = _extract_at_agent(arg, "")
        return {"action": "cost", "agent_name": agt, **base}

    # ── Secrets & Variables ──
    if cmd in ("/secrets", "/list-secrets"):
        return {"action": "list_secrets", **base}

    if cmd == "/add-secret":
        p = arg.split(None, 1)
        return {"action": "add_secret", "name": p[0] if p else "",
                "value": p[1] if len(p) > 1 else "", **base}

    if cmd in ("/variables", "/vars", "/list-variables"):
        return {"action": "list_variables", **base}

    if cmd in ("/add-variable", "/add-var"):
        p = arg.split(None, 1)
        return {"action": "add_variable", "name": p[0] if p else "",
                "value": p[1] if len(p) > 1 else "", **base}

    # ── Scheduling ──
    if cmd == "/schedules":
        return _parse_schedules_command(arg, base)

    if cmd == "/autoconv":
        return _parse_autoconv_command(arg, base, agent_name)

    # ── Files ──
    if cmd == "/run":
        _run_arg = arg.strip()
        _run_svc = ""
        if _run_arg.startswith("@"):
            _parts = _run_arg.split(None, 1)
            _run_svc = _parts[0][1:]  # strip @
            _run_arg = _parts[1] if len(_parts) > 1 else ""
        return {"action": "exec_inline", "command": _run_arg, "service": _run_svc, **base}

    if cmd == "/diff":
        return {"action": "exec_inline", "command": "git diff", **base}

    if cmd == "/view":
        return {"action": "view_file", "path": arg.strip(), **base}

    # ── Activation ──
    if cmd == "/activate":
        p = arg.split(None, 1)
        return {"action": "activate_resource", "type": p[0] if p else "",
                "name": p[1] if len(p) > 1 else "", **base}

    if cmd == "/deactivate":
        p = arg.split(None, 1)
        return {"action": "deactivate_resource", "type": p[0] if p else "",
                "name": p[1] if len(p) > 1 else "", **base}

    if cmd == "/share":
        p = arg.split()
        return {"action": "share_resource",
                "type": p[0] if len(p) > 0 else "",
                "name": p[1] if len(p) > 1 else "",
                "target": p[2] if len(p) > 2 else "",
                **base}

    if cmd == "/link":
        p = arg.split(None, 1)
        return {"action": "link_account", "provider": p[0] if p else "",
                "external_id": p[1] if len(p) > 1 else "", **base}

    # ── Session ──
    if cmd == "/login":
        return {"action": "login", **base}

    if cmd == "/help":
        topic = arg.strip().lstrip("/")
        return {"action": "help", "topic": topic, **base}

    if cmd == "/doctor":
        return {"action": "doctor", **base}

    if cmd == "/add-dir":
        return {"action": "add_dir", "path": arg.strip(), **base}

    # ── Developer ──
    if cmd == "/install":
        return {"action": "install_tool", "source": arg.strip(), **base}

    if cmd == "/uninstall":
        return {"action": "uninstall_tool", "name": arg.strip(), **base}

    if cmd == "/batch":
        return {"action": "batch", "instruction": arg, **base}

    if cmd == "/debug":
        return {"action": "debug", **base}

    if cmd == "/clear-store":
        return {"action": "clear_store", "target": arg.strip() or "", **base}

    if cmd == "/hooks":
        return _parse_hooks_command(arg, base)

    if cmd == "/plan":
        val = arg.strip().lower()
        if val == "off":
            return {"action": "set_plan_mode", "enabled": False, **base}
        if val == "status":
            return {"action": "get_plan_mode", **base}
        # on or toggle
        return {"action": "set_plan_mode", "enabled": True, **base}

    # ── Media ──
    if cmd == "/imgservice":
        return _parse_media_service_command(arg, base, "image")

    if cmd == "/vidservice":
        return _parse_media_service_command(arg, base, "video")

    # ── Stats & Analysis ──
    if cmd == "/stats":
        return {"action": "stats", **base}

    if cmd == "/pr-comments":
        return {"action": "pr_comments", "pr": arg.strip(), **base}

    if cmd == "/security-review":
        return {"action": "security_review", **base}

    if cmd == "/insights":
        return {"action": "insights", **base}

    if cmd in ("/feedback", "/bug"):
        return {"action": "feedback", "report": arg, **base}

    # ── Server workspace ──
    if cmd == "/workspace":
        sub = arg.strip().lower()
        if sub == "destroy":
            return {"action": "destroy_server_workspace", **base}
        if sub == "status":
            return {"action": "server_workspace_status", **base}
        # Default: create (idempotent)
        return {"action": "create_server_workspace", **base}

    # ── Relay bindings ──
    if cmd == "/relay":
        p = arg.strip().split(None, 1)
        sub = (p[0] if p else "").lower()
        rest = p[1].strip() if len(p) > 1 else ""
        if sub == "link":
            return {"action": "relay_link", "relay_id": rest, **base}
        if sub == "unlink":
            return {"action": "relay_unlink", "relay_id": rest, **base}
        if sub == "default":
            return {"action": "relay_default", "relay_id": rest, **base}
        if sub == "list":
            return {"action": "relay_list_available", **base}
        if sub == "local":
            # /relay local <relay_id> true|false [@agent]
            parts = rest.split()
            if len(parts) < 2:
                return {"action": "relay_set_local", "error": "Usage: /relay local <relay_id> true|false [@agent]", **base}
            rid = parts[0]
            val = parts[1].lower() in ("true", "1", "on", "yes")
            agent = parts[2].lstrip("@") if len(parts) > 2 else ""
            return {"action": "relay_set_local", "relay_id": rid, "local": val, "agent": agent, **base}
        # Default: show linked relays
        return {"action": "relay_status", **base}

    # ── Conversation remote filesystem mounts ──
    if cmd in ("/remote-fs", "/remotefs"):
        p = arg.strip().split(None, 1)
        sub = (p[0] if p else "").lower()
        rest = p[1].strip() if len(p) > 1 else ""
        if sub == "link":
            parts = rest.split()
            return {
                "action": "remote_fs_link",
                "service_id": parts[0] if parts else "",
                "scope": parts[1] if len(parts) > 1 else "",
                **base,
            }
        if sub == "unlink":
            return {"action": "remote_fs_unlink", "service_id": rest, **base}
        if sub == "list":
            return {"action": "remote_fs_list_available", **base}
        return {"action": "remote_fs_status", **base}

    # ── Client-only (not handled server-side) ──
    if cmd in ("/clear", "/clear-ui", "/connect", "/disconnect", "/exit", "/quit",
               "/copy", "/paste", "/upload", "/files"):
        return {"_client_only": True, "command": cmd, "arg": arg}

    # Unknown command — not recognized
    return None


# ── Sub-parsers ───────────────────────────────────────────────────────

def _parse_agent_command(arg: str, base: dict, agent_name: str) -> dict:
    """Parse /agent subcommands. @agent convention for all agent params."""
    p = arg.split(None, 1)
    subcmd = (p[0] if p else "list").lower()
    rest = p[1] if len(p) > 1 else ""

    if subcmd == "list":
        return {"action": "list_agents", **base}
    if subcmd == "create":
        return {"action": "create_agent_interactive", **base}
    if subcmd == "select":
        agt, _ = _extract_at_agent(rest, "")
        return {"action": "select_agent", "agent_name": agt, "name": agt, **base}
    if subcmd == "delete":
        agt, _ = _extract_at_agent(rest, "")
        return {"action": "delete_agent", "agent_name": agt, **base}
    if subcmd == "msg":
        target, msg = _extract_at_agent(rest, agent_name)
        if target.upper() == "ALL":
            return {"action": "broadcast_agents", "message": msg, **base}
        return {"action": "agent_msg", "target_agent": target, "message": msg, **base}
    if subcmd == "interrupt":
        agt, _ = _extract_at_agent(rest, agent_name)
        return {"action": "interrupt", "agent_name": agt, **base}
    if subcmd == "btw":
        target, question = _extract_at_agent(rest, agent_name)
        return {"action": "btw", "agent_name": target, "question": question, **base}
    if subcmd == "resume":
        agt, _ = _extract_at_agent(rest, agent_name)
        return {"action": "resume_agent", "agent_name": agt, **base}
    if subcmd == "setname":
        real, nick = _extract_at_agent(rest, agent_name)
        return {"action": "set_nickname", "agent_name": real, "nickname": nick.strip(),
                **base}
    # Unknown subcommand — treat as select (supports @agent or plain name)
    agt, _ = _extract_at_agent(arg, subcmd)
    return {"action": "select_agent", "agent_name": agt, **base}


def _parse_skill_sugar_command(text: str, base: dict, agent_name: str = "") -> dict:
    """Parse //skill-name [@agent] [args...] as /skill run syntax."""
    rest = text[2:].strip()
    parts = rest.split(None, 2)
    skill_name = parts[0].lstrip("@") if parts else ""
    if skill_name.startswith("/"):
        skill_name = ""
    target = agent_name
    arguments = ""
    if len(parts) > 1:
        if parts[1].startswith("@"):
            target = parts[1].lstrip("@")
            arguments = parts[2] if len(parts) > 2 else ""
        else:
            arguments = parts[1]
            if len(parts) > 2:
                arguments += " " + parts[2]
    return {
        "action": "run_skill",
        "target_agent": target,
        "skill_name": skill_name,
        "arguments": arguments,
        **base,
    }


def _skill_short_description(body: str) -> str:
    """Derive a concise manifest description from a skill body.

    `/skill add` only takes a single free-text argument for the body, so the
    description (injected into the system prompt manifest) is taken from the
    first non-empty line rather than the whole body.
    """
    for line in str(body or "").splitlines():
        line = line.strip().lstrip("#").strip()
        if line:
            return line[:200]
    return ""


def _parse_skill_command(arg: str, base: dict, agent_name: str = "") -> dict:
    p = arg.split(None, 2)
    subcmd = p[0] if p else "list"
    if subcmd == "list":
        return {"action": "list_skills", **base}
    if subcmd == "add":
        force, rest = _parse_leading_force(arg[len(subcmd):].strip())
        parts = rest.split(None, 1)
        body = parts[1] if len(parts) > 1 else ""
        return {"action": "create_skill", "name": parts[0].lstrip("@") if parts else "",
                "description": _skill_short_description(body),
                "instructions": body, "force": force, **base}
    if subcmd in ("update", "modify", "edit"):
        force, rest = _parse_leading_force(arg[len(subcmd):].strip())
        parts = rest.split(None, 1)
        body = parts[1] if len(parts) > 1 else ""
        return {"action": "update_skill", "name": parts[0].lstrip("@") if parts else "",
                "description": _skill_short_description(body),
                "instructions": body, "force": force, **base}
    if subcmd == "del":
        return {"action": "delete_skill", "name": p[1].lstrip("@") if len(p) > 1 else "",
                **base}
    if subcmd == "assign":
        return {
            "action": "assign_skill",
            "agent_name": p[1].lstrip("@") if len(p) > 1 else "",
            "skill_name": p[2].split(None, 1)[0].lstrip("@") if len(p) > 2 else "",
            **base,
        }
    if subcmd == "unassign":
        return {
            "action": "unassign_skill",
            "agent_name": p[1].lstrip("@") if len(p) > 1 else "",
            "skill_name": p[2].split(None, 1)[0].lstrip("@") if len(p) > 2 else "",
            **base,
        }
    if subcmd == "assigned":
        return {
            "action": "list_agent_skills",
            "agent_name": p[1].lstrip("@") if len(p) > 1 else "",
            **base,
        }
    if subcmd == "search":
        rest = arg[len(subcmd):].strip()
        try:
            tokens = shlex.split(rest)
        except ValueError:
            tokens = rest.split()
        source = "all"
        query_parts = []
        i = 0
        while i < len(tokens):
            tok = tokens[i]
            if tok == "--source" and i + 1 < len(tokens):
                source = tokens[i + 1]
                i += 2
                continue
            query_parts.append(tok)
            i += 1
        return {
            "action": "search_skill_marketplace",
            "source": source,
            "query": " ".join(query_parts).strip(),
            **base,
        }
    if subcmd == "import":
        rest = arg[len(subcmd):].strip()
        try:
            tokens = shlex.split(rest)
        except ValueError as exc:
            return {"action": "import_skill_marketplace", "ref": "", "error": str(exc), **base}
        source = ""
        review_only = False
        force = False
        scope = "user"
        name = ""
        ref_parts = []
        i = 0
        while i < len(tokens):
            tok = tokens[i]
            if tok == "--source" and i + 1 < len(tokens):
                source = tokens[i + 1]
                i += 2
                continue
            if tok == "--review-only":
                review_only = True
                i += 1
                continue
            if tok == "--force":
                force = True
                i += 1
                continue
            if tok == "--scope" and i + 1 < len(tokens):
                scope = tokens[i + 1]
                i += 2
                continue
            if tok == "--name" and i + 1 < len(tokens):
                name = tokens[i + 1]
                i += 2
                continue
            ref_parts.append(tok)
            i += 1
        return {
            "action": "import_skill_marketplace",
            "source": source,
            "ref": " ".join(ref_parts).strip(),
            "name": name,
            "review_only": review_only,
            "force": force,
            "scope": scope,
            **base,
        }
    if subcmd == "run":
        rest = arg[len(subcmd):].strip()
        target, rest = _extract_at_agent(rest, agent_name)
        parts = rest.split(None, 1)
        skill_name = parts[0].lstrip("@") if parts else ""
        arguments = parts[1] if len(parts) > 1 else ""
        return {
            "action": "run_skill",
            "target_agent": target,
            "skill_name": skill_name,
            "arguments": arguments,
            **base,
        }
    return {"action": "list_skills", **base}


def _parse_leading_force(rest: str) -> tuple[bool, str]:
    rest = (rest or "").strip()
    if rest == "--force":
        return True, ""
    if rest.startswith("--force "):
        return True, rest[len("--force "):].strip()
    return False, rest


def _parse_pfp_command(arg: str, base: dict) -> dict:
    try:
        tokens = shlex.split(arg or "")
    except ValueError as exc:
        return {"action": "pfp_error", "error": str(exc), **base}
    subcmd = tokens[0].lower() if tokens else "list"
    rest = tokens[1:]
    result = {"action": f"pfp_{subcmd.replace('-', '_')}", **base}
    if subcmd in ("key-create", "key_create"):
        result["action"] = "pfp_key_create"
        return result
    if subcmd in ("list", "list-installed", "list_installed"):
        result["action"] = "pfp_list_installed"
        result.update(_parse_pfp_flags(rest))
        return result
    if subcmd in ("reload-tasks", "reload_tasks"):
        result["action"] = "pfp_reload_tasks"
        result.update(_parse_pfp_flags(rest))
        return result
    if subcmd == "search":
        flags = _parse_pfp_flags(rest)
        result.update(flags)
        result["query"] = " ".join(flags.get("_positionals", [])).strip()
        return result
    if subcmd == "registry":
        return _parse_pfp_registry_command(rest, base)
    if subcmd == "inspect":
        flags = _parse_pfp_flags(rest)
        result.update(flags)
        result["path"] = flags.get("path", "")
        return result
    if subcmd == "build":
        flags = _parse_pfp_flags(rest)
        result.update(flags)
        result["source_dir"] = flags.get("path", "")
        return result
    if subcmd == "install":
        flags = _parse_pfp_flags(rest)
        result.update(flags)
        result["path"] = flags.get("path", "")
        return result
    if subcmd in ("dev-load", "dev_load"):
        flags = _parse_pfp_flags(rest)
        if "--scope" not in rest:
            flags["scope"] = "conversation"
        result["action"] = "pfp_dev_load"
        result.update(flags)
        result["source_dir"] = flags.get("path", "")
        return result
    if subcmd in ("dev-unload", "dev_unload"):
        flags = _parse_pfp_flags(rest)
        if "--scope" not in rest:
            flags["scope"] = "conversation"
        result["action"] = "pfp_dev_unload"
        result.update(flags)
        result["package"] = flags.get("path", "")
        return result
    if subcmd == "update":
        flags = _parse_pfp_flags(rest)
        result.update(flags)
        result["path"] = flags.get("path", "")
        return result
    if subcmd == "uninstall":
        flags = _parse_pfp_flags(rest)
        result.update(flags)
        result["package"] = flags.get("path", "")
        return result
    if subcmd == "export":
        result.update(_parse_pfp_flags(rest))
        return result
    return {"action": "pfp_error", "error": f"Unknown /pfp subcommand: {subcmd}", **base}


def _parse_pfp_flags(tokens: List[str]) -> dict:
    data = {
        "scope": "user",
        "include": [],
        "exclude": [],
        "force": False,
        "replace": False,
        "dry_run": False,
        "path": "",
        "secret_bindings": {},
    }
    positional = []
    i = 0
    while i < len(tokens):
        tok = tokens[i]
        if tok == "--scope" and i + 1 < len(tokens):
            data["scope"] = tokens[i + 1]
            i += 2
            continue
        if tok == "--include" and i + 1 < len(tokens):
            data["include"].extend(_split_csv(tokens[i + 1]))
            i += 2
            continue
        if tok == "--exclude" and i + 1 < len(tokens):
            data["exclude"].extend(_split_csv(tokens[i + 1]))
            i += 2
            continue
        if tok == "--out" and i + 1 < len(tokens):
            data["output_path"] = tokens[i + 1]
            data["output_dir"] = tokens[i + 1]
            i += 2
            continue
        if tok == "--key" and i + 1 < len(tokens):
            data["private_key"] = tokens[i + 1]
            i += 2
            continue
        if tok == "--key-env" and i + 1 < len(tokens):
            data["private_key_env"] = tokens[i + 1]
            i += 2
            continue
        if tok == "--package" and i + 1 < len(tokens):
            data["package"] = tokens[i + 1]
            i += 2
            continue
        if tok == "--version" and i + 1 < len(tokens):
            data["version"] = tokens[i + 1]
            i += 2
            continue
        if tok == "--name" and i + 1 < len(tokens):
            data["name"] = tokens[i + 1]
            i += 2
            continue
        if tok == "--sha256" and i + 1 < len(tokens):
            data["sha256"] = tokens[i + 1]
            i += 2
            continue
        if tok == "--secret" and i + 1 < len(tokens):
            name, _, key = tokens[i + 1].partition("=")
            if name.strip() and key.strip():
                data["secret_bindings"][name.strip()] = key.strip()
            i += 2
            continue
        if tok == "--limit" and i + 1 < len(tokens):
            try:
                data["limit"] = int(tokens[i + 1])
            except ValueError:
                data["limit"] = tokens[i + 1]
            i += 2
            continue
        if tok == "--force":
            data["force"] = True
            i += 1
            continue
        if tok == "--confirm-download":
            data["confirm_download"] = True
            i += 1
            continue
        if tok == "--replace":
            data["replace"] = True
            i += 1
            continue
        if tok == "--trusted":
            data["trusted"] = True
            i += 1
            continue
        if tok == "--dry-run":
            data["dry_run"] = True
            i += 1
            continue
        positional.append(tok)
        i += 1
    if positional and not data.get("path"):
        data["path"] = positional[0]
    data["_positionals"] = positional
    return data


def _split_csv(value: str) -> List[str]:
    return [part.strip() for part in str(value or "").split(",") if part.strip()]


def _parse_pfp_registry_command(tokens: List[str], base: dict) -> dict:
    subcmd = tokens[0].lower() if tokens else "list"
    rest = tokens[1:]
    flags = _parse_pfp_flags(rest)
    if subcmd == "add":
        return {"action": "pfp_registry_add", "url": flags.get("path", ""), **flags, **base}
    if subcmd in ("remove", "rm", "delete", "del"):
        return {"action": "pfp_registry_remove", "name": flags.get("path", ""), **flags, **base}
    if subcmd == "list":
        return {"action": "pfp_registry_list", **flags, **base}
    return {"action": "pfp_error", "error": f"Unknown /pfp registry subcommand: {subcmd}", **base}


def _parse_task_command(arg: str, base: dict) -> dict:
    p = arg.split(None, 1)
    subcmd = p[0] if p else "list"
    rest = p[1] if len(p) > 1 else ""

    if subcmd == "list":
        return {"action": "list_tasks", **base}
    if subcmd in ("create", "assign", "delete", "pause", "resume", "cancel"):
        return {"action": f"task_{subcmd}", "args": rest, **base}
    return {"action": "list_tasks", **base}


def _parse_goal_command(arg: str, base: dict, agent_name: str) -> dict:
    import shlex
    try:
        tokens = shlex.split(arg or "")
    except ValueError as e:
        return {"action": "goal", "prompt": "", "error": str(e), **base}
    target = ""
    prompt_parts = []
    variables = {}
    result = {"action": "goal", **base}
    i = 0
    if tokens and tokens[0].startswith("@"):
        target = tokens[0][1:]
        i = 1
    while i < len(tokens):
        tok = tokens[i]
        if tok == "--criteria" and i + 1 < len(tokens):
            result["criteria"] = tokens[i + 1]; i += 2; continue
        if tok == "--interval" and i + 1 < len(tokens):
            result["interval"] = tokens[i + 1]; i += 2; continue
        if tok == "--verifier" and i + 1 < len(tokens):
            v = tokens[i + 1]
            result["verifier"] = v[1:] if v.startswith("@") else v
            i += 2; continue
        if tok == "--budget" and i + 1 < len(tokens):
            result["max_budget"] = tokens[i + 1]; i += 2; continue
        if tok == "--turn-time" and i + 1 < len(tokens):
            result["max_turn_time"] = tokens[i + 1]; i += 2; continue
        if tok == "--total-time" and i + 1 < len(tokens):
            result["max_total_time"] = tokens[i + 1]; i += 2; continue
        if tok == "--max-reschedules" and i + 1 < len(tokens):
            try:
                result["max_reschedules"] = int(tokens[i + 1])
            except ValueError:
                result["max_reschedules"] = 0
            i += 2; continue
        if tok == "--max" and i + 1 < len(tokens):
            try:
                result["max_iterations"] = int(tokens[i + 1])
            except ValueError:
                result["max_iterations"] = 0
            i += 2; continue
        if tok == "--context" and i + 1 < len(tokens):
            result["context"] = tokens[i + 1]; i += 2; continue
        if tok == "--var" and i + 1 < len(tokens):
            kv = tokens[i + 1]
            if "=" in kv:
                k, v = kv.split("=", 1)
                if k:
                    variables[k] = v
            i += 2; continue
        if tok == "--auto-allow":
            result["auto_allow"] = True; i += 1; continue
        if tok == "--interactive":
            result["interactive"] = True; i += 1; continue
        prompt_parts.append(tok)
        i += 1
    result["agent_name"] = target or agent_name
    result["prompt"] = " ".join(prompt_parts).strip()
    if variables:
        result["variables"] = variables
    return result


def _parse_service_command(arg: str, base: dict, user_id: str) -> dict:
    p = arg.split(None, 1)
    subcmd = p[0] if p else "list"
    if subcmd == "list":
        return {"action": "list_services", **base}
    return {"action": "service_command", "subcommand": subcmd,
            "args": p[1] if len(p) > 1 else "", **base}


def _parse_flow_command(arg: str, base: dict) -> dict:
    p = arg.split(None, 1)
    subcmd = p[0] if p else "list"
    return {"action": "flow_command", "subcommand": subcmd,
            "args": p[1] if len(p) > 1 else "", **base}


def _parse_memory_command(arg: str, base: dict, agent_name: str) -> dict:
    p = arg.split(None, 2)
    subcmd = p[0] if p else "list"
    if subcmd == "list":
        return {"action": "list_memories",
                "agent_name": p[1] if len(p) > 1 else agent_name, **base}
    if subcmd == "add":
        return {"action": "add_memory", "text": p[1] if len(p) > 1 else "",
                "agent_name": agent_name, **base}
    if subcmd == "search":
        return {"action": "search_memories", "query": p[1] if len(p) > 1 else "",
                "agent_name": agent_name, **base}
    if subcmd == "del":
        return {"action": "delete_memory", "memory_id": p[1] if len(p) > 1 else "",
                **base}
    if subcmd == "edit":
        return {"action": "edit_memory",
                "memory_id": p[1] if len(p) > 1 else "",
                "text": p[2] if len(p) > 2 else "", **base}
    return {"action": "list_memories", "agent_name": agent_name, **base}


def _parse_schedules_command(arg: str, base: dict) -> dict:
    p = arg.split(None, 1)
    subcmd = p[0] if p else "list"
    if subcmd == "list":
        return {"action": "list_schedules", **base}
    if subcmd == "add":
        rest = p[1] if len(p) > 1 else ""
        # Parse: <datetime> [reason words...] [@agent] [--loop N]
        parts = rest.split()
        at_str = parts[0] if parts else ""
        agent = ""
        loop_seconds = 0
        reason_parts = []
        i = 1
        while i < len(parts):
            if parts[i].startswith("@"):
                agent = parts[i][1:]  # strip @
                i += 1
            elif parts[i] == "--loop" and i + 1 < len(parts):
                try:
                    loop_seconds = int(parts[i + 1])
                except ValueError:
                    pass
                i += 2
            else:
                reason_parts.append(parts[i])
                i += 1
        reason = " ".join(reason_parts) if reason_parts else "manual schedule"
        result = {"action": "add_schedule", "at": at_str, "reason": reason, **base}
        if agent:
            result["agent"] = agent
        if loop_seconds > 0:
            result["loop_seconds"] = loop_seconds
        return result
    if subcmd == "del":
        key = p[1].strip() if len(p) > 1 else ""
        return {"action": "delete_schedule", "key": key, **base}
    return {"action": "list_schedules", **base}


def _parse_autoconv_command(arg: str, base: dict, agent_name: str) -> dict:
    p = arg.split()
    agt = p[0] if p else agent_name
    subcmd = p[1] if len(p) > 1 else "status"
    if subcmd in ("on", "enable"):
        min_iv = int(p[2]) if len(p) > 2 else 60
        max_iv = int(p[3]) if len(p) > 3 else min_iv * 4
        return {"action": "autoconv", "agent_name": agt, "enabled": True,
                "min_interval": min_iv, "max_interval": max_iv, **base}
    if subcmd in ("off", "disable"):
        return {"action": "autoconv", "agent_name": agt, "enabled": False,
                **base}
    return {"action": "autoconv_status", "agent_name": agt, **base}


def _parse_media_service_command(arg: str, base: dict, media_type: str) -> dict:
    p = arg.split()
    subcmd = p[0] if p else "list"
    if subcmd == "list":
        return {"action": f"list_{media_type}_services", **base}
    if subcmd == "select":
        svc = p[1] if len(p) > 1 else ""
        agt = p[2] if len(p) > 2 else "*"
        return {"action": f"set_{media_type}_service",
                "service_name": svc, "agent_name": agt, **base}
    if subcmd == "clear":
        agt = p[1] if len(p) > 1 else "*"
        return {"action": f"clear_{media_type}_service", "agent_name": agt,
                **base}
    return {"action": f"list_{media_type}_services", **base}


def _parse_hooks_command(arg: str, base: dict) -> dict:
    p = arg.split(None, 2)
    subcmd = p[0] if p else "list"
    if subcmd == "list":
        return {"action": "list_hooks", **base}
    if subcmd == "add":
        return {"action": "add_hook", "spec": p[1] if len(p) > 1 else "",
                "command": p[2] if len(p) > 2 else "", **base}
    if subcmd == "del":
        return {"action": "delete_hook", "hook_id": p[1] if len(p) > 1 else "",
                **base}
    return {"action": "list_hooks", **base}


# ── Main handler ──────────────────────────────────────────────────────

def _handle_command_dispatch(self, action, body, store, user_id, flowfile):
    """Handle the unified 'command' action.

    Receives {action: "command", text: "/compact grok", conversation_id: "..."}
    Parses the text, and either:
    - Returns a result directly (for /help, etc.)
    - Re-dispatches to existing action handlers
    """
    if action != "command":
        return None

    text = body.get("text", "").strip()
    conversation_id = body.get("conversation_id", "")
    agent_name = body.get("agent_name", "")

    if not text.startswith("/"):
        return None

    # Parse the command
    parsed = _parse_command(text, conversation_id, user_id, agent_name)

    if parsed is None:
        # Unknown command
        flowfile.set_content(json.dumps({
            "error": f"Unknown command: {text.split()[0]}",
            "hint": "Type /help to see available commands.",
        }).encode())
        flowfile.set_attribute("http.response.status", "400")
        return [flowfile]

    if parsed.get("_client_only"):
        # Client-only command — tell client to handle it
        flowfile.set_content(json.dumps({
            "client_only": True,
            "command": parsed["command"],
            "arg": parsed.get("arg", ""),
        }).encode())
        return [flowfile]

    # Handle /help directly
    if parsed.get("action") == "help":
        return _handle_help(parsed.get("topic", ""), flowfile)

    # Re-dispatch: inject parsed action into body and let normal handlers process it
    parsed_action = parsed.pop("action")
    # Merge parsed fields into body for the handler
    dispatch_body = {**body, **parsed, "action": parsed_action}

    # Update flowfile with new body
    flowfile.set_content(json.dumps(dispatch_body).encode())

    # Return None to let the normal action dispatch handle it
    # The caller should re-dispatch with the new action
    return {"_redispatch": True, "body": dispatch_body, "flowfile": flowfile}


def _handle_help(topic: str, flowfile: FlowFile) -> list:
    """Handle /help [topic] — return help text."""
    if not topic:
        # List all commands grouped by category
        categories = {
            "Conversation": ["/new", "/conv", "/delete", "/rename", "/export",
                             "/history", "/search", "/fork"],
            "Agent": ["/agent", "/msg", "/btw", "/stop", "/resume", "/setname"],
            "Context": ["/compact", "/git-prune", "/context", "/model", "/llm", "/effort",
                        "/fast", "/rebuild", "/restart", "/rewind", "/summary",
                        "/cc_restart", "/cc_live",
                        "/codex_restart", "/codex_live",
                        "/gemini_restart", "/gemini_live"],
            "Resources": ["/resources", "/tools", "/call", "/skill", "/task",
                          "/service", "/flow", "/prompt", "/memory", "/cost"],
            "Secrets & Variables": ["/secrets", "/add-secret", "/variables",
                                    "/add-variable"],
            "Scheduling": ["/schedules", "/autoconv", "/loop"],
            "Files": ["/files", "/upload", "/paste", "/copy", "/view",
                      "/run", "/diff", "/relay", "/workspace"],
            "Mode": ["/plan", "/hooks", "/permission"],
            "Activation": ["/activate", "/deactivate", "/share", "/link"],
            "Session": ["/login", "/help", "/doctor", "/add-dir"],
            "Analysis": ["/stats", "/insights", "/pr-comments",
                         "/security-review", "/feedback"],
            "Developer": ["/install", "/uninstall", "/batch", "/debug",
                          "/clear-store"],
        }
        lines = []
        for cat, cmds in categories.items():
            lines.append(f"\n**{cat}**")
            for c in cmds:
                info = HELP.get(c, {})
                short = info.get("short", "")
                aliases = info.get("aliases", "")
                alias_str = f" ({aliases})" if aliases else ""
                lines.append(f"  `{c}`{alias_str} — {short}")
        text = "## Available Commands\n" + "\n".join(lines)
        text += "\n\nType `/help <command>` for detailed help."
        flowfile.set_content(json.dumps({"help": text}).encode())
        return [flowfile]

    # Specific command help
    cmd = f"/{topic}" if not topic.startswith("/") else topic
    cmd = ALIASES.get(cmd, cmd)
    info = HELP.get(cmd)
    if not info:
        flowfile.set_content(json.dumps({
            "error": f"No help for '{topic}'",
            "hint": "Type /help to see all commands.",
        }).encode())
        return [flowfile]

    text = f"**{cmd}**"
    if info.get("aliases"):
        text += f" (aliases: {info['aliases']})"
    text += f"\n\nUsage: `{info.get('usage', cmd)}`\n\n{info.get('detail', info.get('short', ''))}"
    flowfile.set_content(json.dumps({"help": text}).encode())
    return [flowfile]
