-
Notifications
You must be signed in to change notification settings - Fork 9
Expand file tree
/
Copy pathcodec_slash_commands.py
More file actions
415 lines (352 loc) Β· 13.7 KB
/
codec_slash_commands.py
File metadata and controls
415 lines (352 loc) Β· 13.7 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
"""CODEC Slash Commands β first-class CLI controls in chat.
Type `/<command>` in chat to invoke meta-commands without an LLM round-trip.
This module is fully ADDITIVE: if no slash is detected, the regular chat flow
runs untouched.
Architecture
------------
- One registry: SLASH_COMMANDS (list of SlashCommand objects)
- One parser: parse_slash() β returns (cmd, args) or None
- One dispatcher: dispatch() β runs the command, returns markdown reply
The dashboard chat handler calls parse_slash() before LLM dispatch. If a
slash is matched, dispatch() runs synchronously and the markdown reply is
streamed back as if it were an LLM response (same SSE shape).
Adding new commands
-------------------
Just append to SLASH_COMMANDS at the bottom of this file. Each handler
takes the parsed args list, returns markdown.
"""
from __future__ import annotations
import json
import os
import platform
import sqlite3
import subprocess
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, Optional
# ββ Paths ββ
CONFIG_PATH = Path(os.path.expanduser("~/.codec/config.json"))
LICENSE_DB = Path(os.path.expanduser("~/ava-stack/license-server/licenses.db"))
# ββ Data model ββ
@dataclass
class SlashCommand:
"""A single slash command registration."""
name: str # e.g. "skills"
handler: Callable[[list[str]], str] # takes args, returns markdown
summary: str # one-line help
usage: str = "" # e.g. "/skills [enable|disable] <name>"
aliases: list[str] = field(default_factory=list)
# ββ Parser ββ
def parse_slash(text: str) -> Optional[tuple[str, list[str]]]:
"""Detect and parse a slash command. Returns (cmd_name, args) or None.
Rules:
- Must start with `/` (after leading whitespace)
- Command name is `[a-zA-Z0-9_-]+`
- Args are space-separated; quoted args preserved
- Backslash-escaped slash `\\/` is NOT treated as slash (passes through)
- Empty `/` or `/ ` is None (not a command)
"""
if not text:
return None
s = text.strip()
if not s.startswith("/"):
return None
if s.startswith("\\/"):
return None
body = s[1:].strip()
if not body:
return None
# Split on whitespace, preserve quoted args
import shlex
try:
parts = shlex.split(body)
except ValueError:
# Unbalanced quote β fall back to simple split
parts = body.split()
if not parts:
return None
name = parts[0].lower().strip()
# Validate the command name (don't false-positive on URLs or paths).
# Allow `?` as a special-case for the /? help alias.
if not all(c.isalnum() or c in "_-?" for c in name):
return None
return name, parts[1:]
# ββ Dispatcher ββ
def find_command(name: str) -> Optional[SlashCommand]:
"""Look up by primary name or alias."""
name = name.lower()
for cmd in SLASH_COMMANDS:
if cmd.name == name or name in cmd.aliases:
return cmd
return None
def dispatch(name: str, args: list[str]) -> str:
"""Run a slash command. Returns markdown reply."""
cmd = find_command(name)
if not cmd:
return _unknown_command(name)
try:
return cmd.handler(args)
except Exception as e:
return f"β οΈ `/{name}` failed: `{type(e).__name__}: {e}`"
def _unknown_command(name: str) -> str:
suggestions = [c.name for c in SLASH_COMMANDS if c.name.startswith(name[:2])]
body = [f"β Unknown slash command: `/{name}`"]
if suggestions:
body.append(f"Did you mean: {', '.join(f'`/{s}`' for s in suggestions[:5])}?")
body.append("Type `/help` for the full list.")
return "\n\n".join(body)
# ββ Helpers ββ
def _load_config() -> dict:
try:
return json.loads(CONFIG_PATH.read_text())
except Exception:
return {}
def _save_config(cfg: dict) -> None:
"""Atomic write to avoid half-written JSON."""
tmp = CONFIG_PATH.with_suffix(".json.tmp")
tmp.write_text(json.dumps(cfg, indent=2))
tmp.replace(CONFIG_PATH)
def _table(headers: list[str], rows: list[list[str]]) -> str:
"""Render a markdown table."""
out = ["| " + " | ".join(headers) + " |"]
out.append("| " + " | ".join("---" for _ in headers) + " |")
for row in rows:
out.append("| " + " | ".join(str(c) for c in row) + " |")
return "\n".join(out)
# ββ Built-in command handlers ββ
def _cmd_help(args: list[str]) -> str:
rows = [[f"`/{c.name}`", c.summary] for c in SLASH_COMMANDS]
return "## CODEC Slash Commands\n\n" + _table(["Command", "Description"], rows) + (
"\n\nType `/help <command>` for usage details on a specific command."
if args else ""
)
def _cmd_skills(args: list[str]) -> str:
"""List, enable, disable, or describe skills."""
# Lazy import to avoid pulling skill registry at module load
try:
from codec_skill_registry import SkillRegistry
from codec_config import SKILLS_DIR
except Exception as e:
return f"β οΈ skill registry unavailable: {e}"
reg = SkillRegistry(SKILLS_DIR)
reg.scan()
cfg = _load_config()
enabled = set(cfg.get("skills", []))
if not args or args[0].lower() in ("list", "ls"):
names = sorted(reg.names())
rows = []
for n in names:
meta = reg.get_meta(n) or {}
on = "β
" if n in enabled or len(enabled) == 0 else "βͺ"
desc = (meta.get("SKILL_DESCRIPTION") or "")[:60]
rows.append([on, f"`{n}`", desc])
return f"## Skills ({len(names)} total)\n\n" + _table(
["", "Name", "Description"], rows
)
sub = args[0].lower()
if sub in ("enable", "on") and len(args) >= 2:
target = args[1]
if target not in reg.names():
return f"β οΈ unknown skill `{target}`"
if "skills" not in cfg or not isinstance(cfg["skills"], list):
cfg["skills"] = list(reg.names())
if target not in cfg["skills"]:
cfg["skills"].append(target)
_save_config(cfg)
return f"β
Skill `{target}` enabled."
if sub in ("disable", "off") and len(args) >= 2:
target = args[1]
cfg["skills"] = [s for s in cfg.get("skills", []) if s != target]
_save_config(cfg)
return f"π« Skill `{target}` disabled."
if sub == "info" and len(args) >= 2:
target = args[1]
meta = reg.get_meta(target)
if not meta:
return f"β οΈ unknown skill `{target}`"
triggers = meta.get("SKILL_TRIGGERS", [])
return (
f"## `{target}`\n\n"
f"**Description:** {meta.get('SKILL_DESCRIPTION', '(none)')}\n\n"
f"**MCP exposed:** {meta.get('SKILL_MCP_EXPOSE', False)}\n\n"
f"**Triggers:** {', '.join(f'`{t}`' for t in triggers[:10]) or '(none)'}"
)
return "Usage: `/skills [list|enable <name>|disable <name>|info <name>]`"
def _cmd_plugins(args: list[str]) -> str:
"""For now an alias of /skills until the plugin lifecycle ships."""
return _cmd_skills(args)
def _cmd_clear(args: list[str]) -> str:
return (
"π§Ή Chat cleared.\n\n"
"*(The dashboard frontend should hide all prior messages on receiving "
"this command. If they're still visible, refresh the page β the "
"frontend hook for `/clear` is being added.)*"
)
def _cmd_version(args: list[str]) -> str:
cfg = _load_config()
rows = [
["CODEC", _git_short_sha(Path(os.path.expanduser("~/codec-repo")))],
["Python", platform.python_version()],
["macOS", platform.mac_ver()[0] or "(unknown)"],
["LLM model", cfg.get("llm_model", "(not set)")],
["Vision model", cfg.get("vision_model", "(not set)")],
["TTS engine", cfg.get("tts_engine", "(not set)")],
["AVA proxy", (cfg.get("ava") or {}).get("proxy_url", "(disabled)")],
["AVA license", _ava_license_status(cfg)],
]
return "## CODEC Version\n\n" + _table(["Component", "Value"], rows)
def _git_short_sha(repo_dir: Path) -> str:
try:
out = subprocess.check_output(
["git", "-C", str(repo_dir), "rev-parse", "--short", "HEAD"],
timeout=2, stderr=subprocess.DEVNULL,
)
return out.decode().strip()
except Exception:
return "(no git)"
def _ava_license_status(cfg: dict) -> str:
ava = cfg.get("ava") or {}
if not ava.get("enabled"):
return "(disabled)"
key = ava.get("license_key", "")
if not key:
return "(no key)"
# Decode JWT payload to show tier+expiry without verification
try:
import base64
parts = key.split(".")
if len(parts) != 3:
return "(malformed)"
payload = parts[1] + "=" * (-len(parts[1]) % 4)
data = json.loads(base64.urlsafe_b64decode(payload))
exp = datetime.fromtimestamp(data.get("exp", 0), tz=timezone.utc)
return f"{data.get('tier', '?')} β’ expires {exp.date().isoformat()}"
except Exception:
return "(unreadable)"
def _cmd_cost(args: list[str]) -> str:
"""Today's spend from the AVA proxy usage table."""
if not LICENSE_DB.exists():
return "β οΈ AVA usage DB not found at `~/ava-stack/license-server/licenses.db`."
today_utc = datetime.now(timezone.utc).date().isoformat()
try:
with sqlite3.connect(str(LICENSE_DB)) as c:
c.row_factory = sqlite3.Row
r = c.execute(
"SELECT COUNT(*) as n, COALESCE(SUM(input_tokens), 0) as in_tok, "
"COALESCE(SUM(output_tokens), 0) as out_tok, "
"COALESCE(SUM(billed_usd_cents), 0) as cents "
"FROM usage WHERE substr(ts,1,10)=?", (today_utc,)
).fetchone()
month_start = datetime.now(timezone.utc).replace(
day=1, hour=0, minute=0, second=0, microsecond=0
).isoformat()
m = c.execute(
"SELECT COUNT(*) as n, COALESCE(SUM(billed_usd_cents), 0) as cents "
"FROM usage WHERE ts >= ?", (month_start,)
).fetchone()
except sqlite3.OperationalError as e:
return f"β οΈ usage DB query failed: {e}"
return (
f"## Today's spend ({today_utc})\n\n"
f"- Requests: **{r['n']}**\n"
f"- Input tokens: **{r['in_tok']:,}**\n"
f"- Output tokens: **{r['out_tok']:,}**\n"
f"- Billed: **${r['cents']/100:.2f}**\n\n"
f"## Month to date\n\n"
f"- Requests: **{m['n']}**\n"
f"- Billed: **${m['cents']/100:.2f}**"
)
def _cmd_status(args: list[str]) -> str:
"""Quick green/red dot for each major service."""
import requests
services = [
("Local Qwen", "http://localhost:8083/v1/models"),
("Whisper STT", "http://localhost:8084/v1/models"),
("Kokoro TTS", "http://localhost:8085/v1/models"),
("AVA proxy", "https://ava-proxy.lucyvpa.com/health"),
("AVA license", "https://ava-license.lucyvpa.com/health"),
]
rows = []
for name, url in services:
t0 = time.monotonic()
try:
r = requests.get(url, timeout=2)
ok = r.status_code < 500
ms = int((time.monotonic() - t0) * 1000)
rows.append([("π’" if ok else "π΄"), name, f"HTTP {r.status_code}", f"{ms}ms"])
except Exception:
rows.append(["π΄", name, "(no response)", "β"])
return "## Service status\n\n" + _table(["", "Service", "HTTP", "Latency"], rows)
def _cmd_who(args: list[str]) -> str:
cfg = _load_config()
return (
f"## CODEC identity\n\n"
f"- Agent name: **{cfg.get('agent_name', 'CODEC')}**\n"
f"- Nickname: **{cfg.get('agent_nickname', '(none)')}**\n"
f"- Wake phrases: {', '.join(f'`{p}`' for p in cfg.get('wake_phrases', []))}\n"
f"- License email: **{(cfg.get('ava') or {}).get('license_key', 'n/a')[:16]}β¦**"
)
# ββ Registry (add new commands here) ββ
SLASH_COMMANDS: list[SlashCommand] = [
SlashCommand(
name="help",
handler=_cmd_help,
summary="List all slash commands",
aliases=["?", "commands"],
),
SlashCommand(
name="skills",
handler=_cmd_skills,
summary="List, enable, disable, or describe skills",
usage="/skills [list|enable <name>|disable <name>|info <name>]",
),
SlashCommand(
name="plugins",
handler=_cmd_plugins,
summary="(alias of /skills until plugin lifecycle ships)",
aliases=["plugin"],
),
SlashCommand(
name="version",
handler=_cmd_version,
summary="Show CODEC version + active models + license",
aliases=["v"],
),
SlashCommand(
name="cost",
handler=_cmd_cost,
summary="Today's and month-to-date AVA proxy spend",
aliases=["spend", "usage"],
),
SlashCommand(
name="status",
handler=_cmd_status,
summary="Quick health check of CODEC's local services",
),
SlashCommand(
name="who",
handler=_cmd_who,
summary="Show CODEC's current persona settings",
),
SlashCommand(
name="clear",
handler=_cmd_clear,
summary="Clear the chat session (frontend hides messages)",
aliases=["cls"],
),
]
# ββ CLI smoke test entry-point ββ
if __name__ == "__main__":
import sys
if len(sys.argv) < 2:
print(_cmd_help([]))
sys.exit(0)
line = " ".join(sys.argv[1:])
parsed = parse_slash(line if line.startswith("/") else "/" + line)
if not parsed:
print(f"Not a slash command: {line!r}")
sys.exit(1)
name, args = parsed
print(dispatch(name, args))