mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-06-10 20:29:00 +08:00
Compare commits
1 Commits
feat/apify
...
hermes/her
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
81d4467052 |
1455
agent/workspace.py
Normal file
1455
agent/workspace.py
Normal file
File diff suppressed because it is too large
Load Diff
3
cli.py
3
cli.py
@@ -4424,6 +4424,9 @@ class HermesCLI:
|
||||
print(f" {status} {p['name']}{version}{detail}{error}")
|
||||
except Exception as e:
|
||||
print(f"Plugin system error: {e}")
|
||||
elif canonical == "workspace":
|
||||
from hermes_cli.workspace import handle_workspace_slash
|
||||
handle_workspace_slash(cmd_original, console=self.console)
|
||||
elif canonical == "rollback":
|
||||
self._handle_rollback_command(cmd_original)
|
||||
elif canonical == "stop":
|
||||
|
||||
@@ -426,6 +426,19 @@ def build_welcome_banner(console: Console, model: str, cwd: str,
|
||||
except Exception:
|
||||
pass # Never break the banner over a profiles.py bug
|
||||
|
||||
# Show active workspace roots
|
||||
try:
|
||||
from agent.workspace import get_workspace_root_specs
|
||||
from hermes_cli.config import load_config as _load_config_for_banner
|
||||
_ws_cfg = _load_config_for_banner()
|
||||
if (_ws_cfg.get("workspace", {}) or {}).get("enabled", True):
|
||||
_ws_roots = get_workspace_root_specs(_ws_cfg)
|
||||
if _ws_roots:
|
||||
_root_labels = [r.label for r in _ws_roots]
|
||||
right_lines.append(f"[bold {accent}]Workspace:[/] [{text}]{', '.join(_root_labels)}[/]")
|
||||
except Exception:
|
||||
pass # Never break the banner over a workspace bug
|
||||
|
||||
right_lines.append(f"[dim {dim}]{' · '.join(summary_parts)}[/]")
|
||||
|
||||
# Update check — use prefetched result if available
|
||||
|
||||
@@ -124,6 +124,10 @@ COMMAND_REGISTRY: list[CommandDef] = [
|
||||
subcommands=("connect", "disconnect", "status")),
|
||||
CommandDef("plugins", "List installed plugins and their status",
|
||||
"Tools & Skills", cli_only=True),
|
||||
CommandDef("workspace", "Workspace status, search, index, and root management",
|
||||
"Tools & Skills", cli_only=True,
|
||||
args_hint="[status|index|list|search|retrieve|roots]",
|
||||
subcommands=("status", "index", "list", "search", "retrieve", "roots")),
|
||||
|
||||
# Info
|
||||
CommandDef("commands", "Browse all commands and skills (paginated)", "Info",
|
||||
|
||||
@@ -187,7 +187,7 @@ def ensure_hermes_home():
|
||||
home = get_hermes_home()
|
||||
home.mkdir(parents=True, exist_ok=True)
|
||||
_secure_dir(home)
|
||||
for subdir in ("cron", "sessions", "logs", "memories"):
|
||||
for subdir in ("cron", "sessions", "logs", "memories", "workspace", "knowledgebase"):
|
||||
d = home / subdir
|
||||
d.mkdir(parents=True, exist_ok=True)
|
||||
_secure_dir(d)
|
||||
@@ -545,8 +545,44 @@ DEFAULT_CONFIG = {
|
||||
"backup_count": 3, # Number of rotated backup files to keep
|
||||
},
|
||||
|
||||
# Workspace — local-first document store for RAG-powered knowledge
|
||||
"workspace": {
|
||||
"enabled": True,
|
||||
"path": "", # empty = HERMES_HOME/workspace
|
||||
},
|
||||
|
||||
# Knowledgebase — indexing, retrieval, and embedding config for workspace RAG
|
||||
"knowledgebase": {
|
||||
"path": "", # empty = HERMES_HOME/knowledgebase
|
||||
"roots": [], # additional directories to index: [{path: "...", recursive: false}]
|
||||
"retrieval_mode": "off", # off | gated | always
|
||||
"auto_index": True, # auto-rebuild index before retrieval
|
||||
"chunking": {
|
||||
"default_tokens": 512,
|
||||
"overlap_tokens": 80,
|
||||
},
|
||||
"embeddings": {
|
||||
"provider": "local", # local | openai | google
|
||||
"model": "google/embeddinggemma-300m",
|
||||
"dimensions": 768,
|
||||
},
|
||||
"reranking": {
|
||||
"provider": "local", # local | cohere | voyage | heuristic
|
||||
"model": "",
|
||||
},
|
||||
"indexing": {
|
||||
"max_file_mb": 10,
|
||||
},
|
||||
"dense_top_k": 40,
|
||||
"sparse_top_k": 40,
|
||||
"fused_top_k": 30,
|
||||
"final_top_k": 8,
|
||||
"max_injected_chunks": 6,
|
||||
"max_injected_tokens": 3200,
|
||||
},
|
||||
|
||||
# Config schema version - bump this when adding new required fields
|
||||
"_config_version": 12,
|
||||
"_config_version": 13,
|
||||
}
|
||||
|
||||
# =============================================================================
|
||||
|
||||
@@ -5485,6 +5485,46 @@ Examples:
|
||||
)
|
||||
logs_parser.set_defaults(func=cmd_logs)
|
||||
|
||||
# =========================================================================
|
||||
# hermes workspace
|
||||
# =========================================================================
|
||||
workspace_parser = subparsers.add_parser(
|
||||
"workspace",
|
||||
help="Manage workspace knowledgebase and RAG",
|
||||
description="Inspect, index, search, and manage workspace roots for RAG retrieval",
|
||||
)
|
||||
workspace_subs = workspace_parser.add_subparsers(dest="workspace_action")
|
||||
workspace_subs.add_parser("status", help="Show workspace status and root configuration")
|
||||
ws_index = workspace_subs.add_parser("index", help="Rebuild chunk index for all workspace roots")
|
||||
ws_list = workspace_subs.add_parser("list", help="List workspace files")
|
||||
ws_list.add_argument("path", nargs="?", default="", help="Subpath to scope listing")
|
||||
ws_list.add_argument("--recursive", action="store_true", default=True, help="Recurse into subdirectories")
|
||||
ws_list.add_argument("--limit", type=int, default=20, help="Max entries to return")
|
||||
ws_list.add_argument("--offset", type=int, default=0, help="Skip first N entries")
|
||||
ws_search = workspace_subs.add_parser("search", help="Search workspace files by regex")
|
||||
ws_search.add_argument("query", help="Regex search query")
|
||||
ws_search.add_argument("--path", default="", help="Subpath to scope search")
|
||||
ws_search.add_argument("--file-glob", dest="file_glob", default=None, help="Filename glob filter, e.g. '*.md'")
|
||||
ws_search.add_argument("--limit", type=int, default=10, help="Max matches to return")
|
||||
ws_search.add_argument("--offset", type=int, default=0, help="Skip first N matches")
|
||||
ws_retrieve = workspace_subs.add_parser("retrieve", help="Hybrid RAG retrieval from workspace index")
|
||||
ws_retrieve.add_argument("query", help="Retrieval query")
|
||||
ws_retrieve.add_argument("--limit", type=int, default=8, help="Max results to return")
|
||||
ws_roots = workspace_subs.add_parser("roots", help="Manage additional workspace roots")
|
||||
ws_roots_subs = ws_roots.add_subparsers(dest="root_action")
|
||||
ws_roots_subs.add_parser("list", help="List active workspace roots")
|
||||
ws_roots_add = ws_roots_subs.add_parser("add", help="Add an additional workspace root")
|
||||
ws_roots_add.add_argument("root_path", help="Directory path to add")
|
||||
ws_roots_add.add_argument("--recursive", action="store_true", default=False, help="Index subdirectories recursively")
|
||||
ws_roots_remove = ws_roots_subs.add_parser("remove", help="Remove a workspace root")
|
||||
ws_roots_remove.add_argument("identifier", help="Path or label of the root to remove")
|
||||
|
||||
def cmd_workspace(args):
|
||||
from hermes_cli.workspace import workspace_command
|
||||
workspace_command(args)
|
||||
|
||||
workspace_parser.set_defaults(func=cmd_workspace)
|
||||
|
||||
# =========================================================================
|
||||
# Parse and execute
|
||||
# =========================================================================
|
||||
|
||||
273
hermes_cli/workspace.py
Normal file
273
hermes_cli/workspace.py
Normal file
@@ -0,0 +1,273 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
|
||||
from rich.console import Console
|
||||
|
||||
from agent.workspace import (
|
||||
add_workspace_root_to_config,
|
||||
index_workspace_knowledgebase,
|
||||
list_workspace_roots,
|
||||
remove_workspace_root_from_config,
|
||||
workspace_list,
|
||||
workspace_retrieve,
|
||||
workspace_search,
|
||||
workspace_status,
|
||||
)
|
||||
from hermes_cli.config import load_config, save_config
|
||||
|
||||
|
||||
def _console(console: Optional[Console]) -> Console:
|
||||
return console or Console()
|
||||
|
||||
|
||||
def _print_status(console: Console) -> None:
|
||||
data = workspace_status(load_config())
|
||||
if not data.get("success"):
|
||||
console.print(f"[bold red]{data.get('error', 'Workspace unavailable')}[/]")
|
||||
return
|
||||
console.print(f"Workspace root: {data['workspace_root']}")
|
||||
console.print(f"Knowledgebase root: {data['knowledgebase_root']}")
|
||||
console.print(f"Manifest: {data['manifest_path']}")
|
||||
console.print(f"Index DB: {data.get('index_path', '(not built)')}")
|
||||
console.print(f"Files: {data['file_count']}")
|
||||
console.print(f"Chunks: {data.get('chunk_count', 0)}")
|
||||
if data.get('embedding_backend'):
|
||||
console.print(f"Embedding backend: {data['embedding_backend']}")
|
||||
if data.get('dense_backend'):
|
||||
console.print(f"Dense backend: {data['dense_backend']}")
|
||||
roots = data.get("active_roots") or []
|
||||
if roots:
|
||||
console.print("Active roots:")
|
||||
for root in roots:
|
||||
mode = "recursive" if root.get("recursive") else "shallow"
|
||||
workspace_tag = " (canonical)" if root.get("is_workspace") else ""
|
||||
console.print(f" - {root['label']}: {root['path']} [{mode}]{workspace_tag}")
|
||||
counts = data.get("category_counts") or {}
|
||||
if counts:
|
||||
for key in sorted(counts):
|
||||
console.print(f" {key}: {counts[key]}")
|
||||
|
||||
|
||||
def _print_index(console: Console) -> None:
|
||||
data = index_workspace_knowledgebase(load_config())
|
||||
if not data.get("success"):
|
||||
console.print(f"[bold red]{data.get('error', 'Index failed')}[/]")
|
||||
return
|
||||
console.print(f"Indexed {data['file_count']} files into {data.get('chunk_count', 0)} chunks")
|
||||
console.print(f"Manifest: {data['manifest_path']}")
|
||||
console.print(f"Index DB: {data['index_path']}")
|
||||
if data.get('embedding_backend'):
|
||||
console.print(f"Embedding backend: {data['embedding_backend']}")
|
||||
if data.get('dense_backend'):
|
||||
console.print(f"Dense backend: {data['dense_backend']}")
|
||||
|
||||
|
||||
def _print_roots(console: Console) -> None:
|
||||
data = list_workspace_roots(load_config())
|
||||
roots = data.get("roots") or []
|
||||
if not roots:
|
||||
console.print("No active workspace roots.")
|
||||
return
|
||||
for root in roots:
|
||||
mode = "recursive" if root.get("recursive") else "shallow"
|
||||
workspace_tag = " (canonical)" if root.get("is_workspace") else ""
|
||||
console.print(f"{root['label']}: {root['path']} ({mode}){workspace_tag}")
|
||||
|
||||
|
||||
def add_workspace_root(root_path: str, recursive: bool = False) -> dict:
|
||||
config = load_config()
|
||||
result = add_workspace_root_to_config(config, root_path, recursive=recursive)
|
||||
if result.get("success"):
|
||||
save_config(config)
|
||||
return result
|
||||
|
||||
|
||||
def remove_workspace_root(identifier: str) -> dict:
|
||||
config = load_config()
|
||||
result = remove_workspace_root_from_config(config, identifier)
|
||||
if result.get("success"):
|
||||
save_config(config)
|
||||
return result
|
||||
|
||||
|
||||
def _print_list(console: Console, path: str = "", recursive: bool = True, limit: int = 20, offset: int = 0) -> None:
|
||||
data = workspace_list(load_config(), relative_path=path, recursive=recursive, limit=limit, offset=offset)
|
||||
if not data.get("success"):
|
||||
console.print(f"[bold red]{data.get('error', 'List failed')}[/]")
|
||||
return
|
||||
entries = data.get("entries") or []
|
||||
if not entries:
|
||||
console.print("No workspace files found.")
|
||||
return
|
||||
for entry in entries:
|
||||
console.print(entry["relative_path"])
|
||||
if data.get("total_count", len(entries)) > len(entries):
|
||||
console.print(f"[dim]Showing {len(entries)} of {data['total_count']} files[/]")
|
||||
|
||||
|
||||
def _print_search(console: Console, query: str, path: str = "", file_glob: str | None = None, limit: int = 10, offset: int = 0) -> None:
|
||||
data = workspace_search(query, load_config(), relative_path=path, file_glob=file_glob, limit=limit, offset=offset)
|
||||
if not data.get("success"):
|
||||
console.print(f"[bold red]{data.get('error', 'Search failed')}[/]")
|
||||
return
|
||||
matches = data.get("matches") or []
|
||||
if not matches:
|
||||
console.print("No matches found.")
|
||||
return
|
||||
for match in matches:
|
||||
console.print(f"{match['relative_path']}:{match['line']} {match['content']}")
|
||||
if data.get("total_count", len(matches)) > len(matches):
|
||||
console.print(f"[dim]Showing {len(matches)} of {data['total_count']} matches[/]")
|
||||
|
||||
|
||||
def _print_retrieve(console: Console, query: str, limit: int = 8) -> None:
|
||||
data = workspace_retrieve(query, load_config(), limit=limit)
|
||||
if not data.get("success"):
|
||||
console.print(f"[bold red]{data.get('error', 'Retrieve failed')}[/]")
|
||||
return
|
||||
results = data.get("results") or []
|
||||
if not results:
|
||||
console.print("No retrieval results found.")
|
||||
return
|
||||
if data.get('dense_backend') or data.get('rerank_backend'):
|
||||
console.print(f"Dense backend: {data.get('dense_backend', '')} Rerank backend: {data.get('rerank_backend', '')}")
|
||||
for result in results:
|
||||
rerank_score = result.get('rerank_score')
|
||||
rerank_text = f" rerank={rerank_score:.3f}" if isinstance(rerank_score, (int, float)) else ""
|
||||
console.print(f"{result['relative_path']} [rrf={result['rrf_score']:.4f} dense={result['dense_score']:.3f}{rerank_text}]")
|
||||
console.print(result["content"])
|
||||
console.print()
|
||||
|
||||
|
||||
def workspace_command(args, console: Optional[Console] = None) -> None:
|
||||
console = _console(console)
|
||||
action = getattr(args, "workspace_action", None) or "status"
|
||||
if action == "status":
|
||||
_print_status(console)
|
||||
elif action == "index":
|
||||
_print_index(console)
|
||||
elif action == "list":
|
||||
_print_list(
|
||||
console,
|
||||
path=getattr(args, "path", "") or "",
|
||||
recursive=getattr(args, "recursive", True),
|
||||
limit=getattr(args, "limit", 20),
|
||||
offset=getattr(args, "offset", 0),
|
||||
)
|
||||
elif action == "search":
|
||||
query = getattr(args, "query", "") or ""
|
||||
if not query.strip():
|
||||
console.print("Usage: hermes workspace search <query>")
|
||||
return
|
||||
_print_search(
|
||||
console,
|
||||
query=query,
|
||||
path=getattr(args, "path", "") or "",
|
||||
file_glob=getattr(args, "file_glob", None),
|
||||
limit=getattr(args, "limit", 10),
|
||||
offset=getattr(args, "offset", 0),
|
||||
)
|
||||
elif action == "retrieve":
|
||||
query = getattr(args, "query", "") or ""
|
||||
if not query.strip():
|
||||
console.print("Usage: hermes workspace retrieve <query>")
|
||||
return
|
||||
_print_retrieve(console, query=query, limit=getattr(args, "limit", 8))
|
||||
elif action == "roots":
|
||||
root_action = getattr(args, "root_action", "list") or "list"
|
||||
if root_action == "list":
|
||||
_print_roots(console)
|
||||
elif root_action == "add":
|
||||
root_path = getattr(args, "root_path", "") or ""
|
||||
if not root_path:
|
||||
console.print("Usage: hermes workspace roots add <path> [--recursive]")
|
||||
return
|
||||
result = add_workspace_root(root_path, recursive=bool(getattr(args, "recursive", False)))
|
||||
if result.get("success"):
|
||||
root = result["root"]
|
||||
mode = "recursive" if root.get("recursive") else "shallow"
|
||||
console.print(f"Added workspace root: {root['path']} ({mode})")
|
||||
else:
|
||||
console.print(f"[bold red]{result.get('error', 'Failed to add root')}[/]")
|
||||
elif root_action == "remove":
|
||||
identifier = getattr(args, "identifier", "") or ""
|
||||
if not identifier:
|
||||
console.print("Usage: hermes workspace roots remove <path-or-label>")
|
||||
return
|
||||
result = remove_workspace_root(identifier)
|
||||
if result.get("success"):
|
||||
console.print(f"Removed workspace root: {result['removed']['path']}")
|
||||
else:
|
||||
console.print(f"[bold red]{result.get('error', 'Failed to remove root')}[/]")
|
||||
else:
|
||||
console.print("Usage: hermes workspace roots [list|add|remove]")
|
||||
else:
|
||||
console.print(f"[bold red]Unknown workspace action: {action}[/]")
|
||||
|
||||
|
||||
def handle_workspace_slash(cmd: str, console: Optional[Console] = None) -> None:
|
||||
console = _console(console)
|
||||
parts = cmd.strip().split()
|
||||
if parts and parts[0].lower() == "/workspace":
|
||||
parts = parts[1:]
|
||||
|
||||
if not parts or parts[0] in {"status", "path"}:
|
||||
_print_status(console)
|
||||
return
|
||||
|
||||
action = parts[0].lower()
|
||||
if action == "index":
|
||||
_print_index(console)
|
||||
return
|
||||
if action == "list":
|
||||
path = parts[1] if len(parts) > 1 else ""
|
||||
_print_list(console, path=path)
|
||||
return
|
||||
if action == "search":
|
||||
query = " ".join(parts[1:]).strip()
|
||||
if not query:
|
||||
console.print("Usage: /workspace search <query>")
|
||||
return
|
||||
_print_search(console, query=query)
|
||||
return
|
||||
if action == "retrieve":
|
||||
query = " ".join(parts[1:]).strip()
|
||||
if not query:
|
||||
console.print("Usage: /workspace retrieve <query>")
|
||||
return
|
||||
_print_retrieve(console, query=query)
|
||||
return
|
||||
if action == "roots":
|
||||
if len(parts) == 1 or parts[1].lower() == "list":
|
||||
_print_roots(console)
|
||||
return
|
||||
sub = parts[1].lower()
|
||||
if sub == "add":
|
||||
if len(parts) < 3:
|
||||
console.print("Usage: /workspace roots add <path> [--recursive]")
|
||||
return
|
||||
recursive = "--recursive" in parts[3:] or "--recursive" in parts[2:]
|
||||
root_path = parts[2]
|
||||
result = add_workspace_root(root_path, recursive=recursive)
|
||||
if result.get("success"):
|
||||
root = result["root"]
|
||||
mode = "recursive" if root.get("recursive") else "shallow"
|
||||
console.print(f"Added workspace root: {root['path']} ({mode})")
|
||||
else:
|
||||
console.print(f"[bold red]{result.get('error', 'Failed to add root')}[/]")
|
||||
return
|
||||
if sub == "remove":
|
||||
if len(parts) < 3:
|
||||
console.print("Usage: /workspace roots remove <path-or-label>")
|
||||
return
|
||||
result = remove_workspace_root(parts[2])
|
||||
if result.get("success"):
|
||||
console.print(f"Removed workspace root: {result['removed']['path']}")
|
||||
else:
|
||||
console.print(f"[bold red]{result.get('error', 'Failed to remove root')}[/]")
|
||||
return
|
||||
console.print("Usage: /workspace roots [list|add|remove]")
|
||||
return
|
||||
|
||||
console.print("Usage: /workspace [status|index|list [path]|search <query>|retrieve <query>|roots ...]")
|
||||
@@ -158,6 +158,7 @@ def _discover_tools():
|
||||
"tools.send_message_tool",
|
||||
# "tools.honcho_tools", # Removed — Honcho is now a memory provider plugin
|
||||
"tools.homeassistant_tool",
|
||||
"tools.workspace_tool",
|
||||
]
|
||||
import importlib
|
||||
for mod_name in _modules:
|
||||
|
||||
@@ -58,6 +58,7 @@ pty = [
|
||||
"pywinpty>=2.0.0,<3; sys_platform == 'win32'",
|
||||
]
|
||||
honcho = ["honcho-ai>=2.0.1,<3"]
|
||||
workspace-rag = ["sentence-transformers>=3.0.0,<4", "torch>=2.0.0", "sqlite-vec>=0.1.0,<1"]
|
||||
mcp = ["mcp>=1.2.0,<2"]
|
||||
homeassistant = ["aiohttp>=3.9.0,<4"]
|
||||
sms = ["aiohttp>=3.9.0,<4"]
|
||||
|
||||
11
run_agent.py
11
run_agent.py
@@ -6907,6 +6907,17 @@ class AIAgent:
|
||||
_should_review_memory = True
|
||||
self._turns_since_memory = 0
|
||||
|
||||
# ── Workspace RAG context (turn-scoped, cache-safe) ──
|
||||
# Appends relevant workspace chunks to this turn's user message only.
|
||||
# Never touches the system prompt or cached prefix.
|
||||
try:
|
||||
from agent.workspace import workspace_context_for_turn
|
||||
_ws_ctx = workspace_context_for_turn(user_message)
|
||||
if _ws_ctx:
|
||||
user_message = user_message + "\n\n" + _ws_ctx
|
||||
except Exception:
|
||||
pass # graceful degradation — workspace not configured or deps missing
|
||||
|
||||
# Add user message
|
||||
user_msg = {"role": "user", "content": user_message}
|
||||
messages.append(user_msg)
|
||||
|
||||
318
tests/agent/test_workspace.py
Normal file
318
tests/agent/test_workspace.py
Normal file
@@ -0,0 +1,318 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
|
||||
def _config(tmp_path: Path) -> dict:
|
||||
return {
|
||||
"workspace": {
|
||||
"enabled": True,
|
||||
"path": str(tmp_path / "workspace"),
|
||||
"auto_create": True,
|
||||
"persist_gateway_uploads": "ask",
|
||||
},
|
||||
"knowledgebase": {
|
||||
"enabled": True,
|
||||
"path": str(tmp_path / "knowledgebase"),
|
||||
"roots": [],
|
||||
"retrieval_mode": "off",
|
||||
"auto_index": True,
|
||||
"watch_for_changes": False,
|
||||
"max_injected_chunks": 6,
|
||||
"max_injected_tokens": 3200,
|
||||
"dense_top_k": 40,
|
||||
"sparse_top_k": 40,
|
||||
"fused_top_k": 30,
|
||||
"final_top_k": 8,
|
||||
"min_fused_score": 0.0,
|
||||
"injection_format": "sourced_note",
|
||||
"chunking": {
|
||||
"default_tokens": 512,
|
||||
"overlap_tokens": 80,
|
||||
"code_strategy": "structural",
|
||||
"markdown_strategy": "headings",
|
||||
},
|
||||
"embeddings": {
|
||||
"provider": "local",
|
||||
"model": "google/embeddinggemma-300m",
|
||||
"dimensions": 768,
|
||||
},
|
||||
"reranker": {
|
||||
"enabled": False,
|
||||
"provider": "local",
|
||||
"model": "bge-reranker-v2-m3",
|
||||
},
|
||||
"indexing": {
|
||||
"respect_gitignore": True,
|
||||
"respect_hermesignore": True,
|
||||
"include_hidden": False,
|
||||
"max_file_mb": 10,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class TestWorkspacePaths:
|
||||
def test_get_workspace_paths_creates_expected_directories(self, tmp_path):
|
||||
from agent.workspace import get_workspace_paths
|
||||
|
||||
paths = get_workspace_paths(_config(tmp_path), ensure=True)
|
||||
|
||||
assert paths.workspace_root == tmp_path / "workspace"
|
||||
assert paths.knowledgebase_root == tmp_path / "knowledgebase"
|
||||
for subdir in ("docs", "notes", "data", "code", "uploads", "media"):
|
||||
assert (paths.workspace_root / subdir).is_dir()
|
||||
assert paths.indexes_dir.is_dir()
|
||||
assert paths.manifests_dir.is_dir()
|
||||
assert paths.cache_dir.is_dir()
|
||||
|
||||
|
||||
class TestWorkspaceManifest:
|
||||
def test_build_workspace_manifest_writes_summary(self, tmp_path):
|
||||
from agent.workspace import build_workspace_manifest
|
||||
|
||||
cfg = _config(tmp_path)
|
||||
workspace = Path(cfg["workspace"]["path"])
|
||||
(workspace / "docs").mkdir(parents=True)
|
||||
(workspace / "notes").mkdir(parents=True)
|
||||
(workspace / "docs" / "a.md").write_text("alpha\n", encoding="utf-8")
|
||||
(workspace / "notes" / "b.txt").write_text("beta\n", encoding="utf-8")
|
||||
|
||||
manifest = build_workspace_manifest(cfg)
|
||||
|
||||
assert manifest["success"] is True
|
||||
assert manifest["file_count"] == 2
|
||||
assert manifest["manifest_path"].endswith("workspace.json")
|
||||
assert Path(manifest["manifest_path"]).exists()
|
||||
paths = {entry["relative_path"] for entry in manifest["files"]}
|
||||
assert paths == {"docs/a.md", "notes/b.txt"}
|
||||
|
||||
saved = json.loads(Path(manifest["manifest_path"]).read_text(encoding="utf-8"))
|
||||
assert saved["file_count"] == 2
|
||||
|
||||
|
||||
class TestWorkspaceSearch:
|
||||
def test_workspace_search_finds_text_matches_and_respects_ignore(self, tmp_path):
|
||||
from agent.workspace import workspace_search
|
||||
|
||||
cfg = _config(tmp_path)
|
||||
workspace = Path(cfg["workspace"]["path"])
|
||||
(workspace / "docs").mkdir(parents=True)
|
||||
(workspace / "docs" / "keep.md").write_text("Hermes likes retrieval\n", encoding="utf-8")
|
||||
(workspace / "docs" / "skip.md").write_text("Hermes hidden\n", encoding="utf-8")
|
||||
(workspace / ".hermesignore").write_text("docs/skip.md\n", encoding="utf-8")
|
||||
(workspace / "docs" / "blob.bin").write_bytes(b"\x00\x01\x02Hermes")
|
||||
|
||||
result = workspace_search("Hermes", config=cfg)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["count"] == 1
|
||||
match = result["matches"][0]
|
||||
assert match["relative_path"] == "docs/keep.md"
|
||||
assert match["line"] == 1
|
||||
|
||||
def test_workspace_search_supports_file_glob(self, tmp_path):
|
||||
from agent.workspace import workspace_search
|
||||
|
||||
cfg = _config(tmp_path)
|
||||
workspace = Path(cfg["workspace"]["path"])
|
||||
(workspace / "docs").mkdir(parents=True)
|
||||
(workspace / "docs" / "a.md").write_text("deploy target\n", encoding="utf-8")
|
||||
(workspace / "docs" / "a.txt").write_text("deploy target\n", encoding="utf-8")
|
||||
|
||||
result = workspace_search("deploy", config=cfg, file_glob="*.md")
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["count"] == 1
|
||||
assert result["matches"][0]["relative_path"] == "docs/a.md"
|
||||
|
||||
|
||||
class TestWorkspaceEmbedder:
|
||||
def test_local_embeddinggemma_uses_sentence_transformers_when_available(self, tmp_path, monkeypatch):
|
||||
from agent.workspace import WorkspaceEmbedder
|
||||
|
||||
calls = {}
|
||||
|
||||
class FakeVector(list):
|
||||
def tolist(self):
|
||||
return list(self)
|
||||
|
||||
class FakeModel:
|
||||
def __init__(self, model_id, **kwargs):
|
||||
calls["model_id"] = model_id
|
||||
calls["kwargs"] = kwargs
|
||||
|
||||
def encode_query(self, text, **kwargs):
|
||||
calls["query"] = (text, kwargs)
|
||||
return FakeVector([0.1, 0.2, 0.3])
|
||||
|
||||
def encode_document(self, texts, **kwargs):
|
||||
calls["documents"] = (list(texts), kwargs)
|
||||
return [FakeVector([0.4, 0.5, 0.6]) for _ in texts]
|
||||
|
||||
fake_torch = SimpleNamespace(
|
||||
cuda=SimpleNamespace(is_available=lambda: False),
|
||||
backends=SimpleNamespace(mps=SimpleNamespace(is_available=lambda: False)),
|
||||
)
|
||||
monkeypatch.setitem(sys.modules, "torch", fake_torch)
|
||||
monkeypatch.setitem(sys.modules, "sentence_transformers", SimpleNamespace(SentenceTransformer=FakeModel))
|
||||
|
||||
embedder = WorkspaceEmbedder(_config(tmp_path))
|
||||
docs = embedder.embed_documents(["alpha doc"])
|
||||
query = embedder.embed_query("alpha query")
|
||||
|
||||
assert embedder.backend == "sentence-transformers"
|
||||
assert calls["model_id"] == "google/embeddinggemma-300m"
|
||||
assert calls["documents"][0] == ["alpha doc"]
|
||||
assert calls["query"][0] == "alpha query"
|
||||
assert docs == [[0.4, 0.5, 0.6]]
|
||||
assert query == [0.1, 0.2, 0.3]
|
||||
|
||||
|
||||
class TestWorkspaceChunking:
|
||||
def test_markdown_chunking_prefers_headings(self, tmp_path):
|
||||
from agent.workspace import _chunk_text
|
||||
|
||||
cfg = _config(tmp_path)
|
||||
text = "# Intro\n\nAlpha overview.\n\n## Deploy\n\nBlue green rollout plan.\n\n## Rollback\n\nRollback steps.\n"
|
||||
chunks = _chunk_text(text, Path("docs/plan.md"), cfg)
|
||||
|
||||
assert len(chunks) >= 3
|
||||
assert any("deploy" in chunk["content"].lower() for chunk in chunks)
|
||||
assert any("rollback" in chunk["content"].lower() for chunk in chunks)
|
||||
|
||||
def test_code_chunking_prefers_symbol_boundaries(self, tmp_path):
|
||||
from agent.workspace import _chunk_text
|
||||
|
||||
cfg = _config(tmp_path)
|
||||
text = "def alpha():\n return 'a'\n\n\ndef beta():\n return 'b'\n"
|
||||
chunks = _chunk_text(text, Path("code/example.py"), cfg)
|
||||
|
||||
assert len(chunks) >= 2
|
||||
assert any("def alpha" in chunk["content"] for chunk in chunks)
|
||||
assert any("def beta" in chunk["content"] for chunk in chunks)
|
||||
|
||||
|
||||
class TestWorkspaceReranker:
|
||||
def test_local_cross_encoder_reranker_reorders_candidates(self, tmp_path, monkeypatch):
|
||||
from agent.workspace import WorkspaceReranker
|
||||
|
||||
calls = {}
|
||||
|
||||
class FakeCrossEncoder:
|
||||
def __init__(self, model_name, **kwargs):
|
||||
calls["model_name"] = model_name
|
||||
calls["kwargs"] = kwargs
|
||||
|
||||
def predict(self, pairs, **kwargs):
|
||||
calls["pairs"] = pairs
|
||||
calls["predict_kwargs"] = kwargs
|
||||
return [0.1, 0.9]
|
||||
|
||||
fake_torch = SimpleNamespace(
|
||||
cuda=SimpleNamespace(is_available=lambda: False),
|
||||
backends=SimpleNamespace(mps=SimpleNamespace(is_available=lambda: False)),
|
||||
)
|
||||
monkeypatch.setitem(sys.modules, "torch", fake_torch)
|
||||
monkeypatch.setitem(sys.modules, "sentence_transformers", SimpleNamespace(CrossEncoder=FakeCrossEncoder))
|
||||
|
||||
cfg = _config(tmp_path)
|
||||
cfg["knowledgebase"]["reranker"]["enabled"] = True
|
||||
cfg["knowledgebase"]["reranker"]["provider"] = "local"
|
||||
cfg["knowledgebase"]["reranker"]["model"] = "cross-encoder/ms-marco-MiniLM-L6-v2"
|
||||
|
||||
reranker = WorkspaceReranker(cfg)
|
||||
ranked = reranker.rerank(
|
||||
"rollback plan",
|
||||
[
|
||||
{"content": "deployment overview", "rrf_score": 0.9, "dense_score": 0.9},
|
||||
{"content": "rollback plan details", "rrf_score": 0.3, "dense_score": 0.2},
|
||||
],
|
||||
)
|
||||
|
||||
assert reranker.backend == "cross-encoder"
|
||||
assert calls["model_name"] == "cross-encoder/ms-marco-MiniLM-L6-v2"
|
||||
assert ranked[0]["content"] == "rollback plan details"
|
||||
|
||||
|
||||
class TestWorkspaceRoots:
|
||||
def test_index_respects_non_recursive_additional_root_by_default(self, tmp_path):
|
||||
from agent.workspace import index_workspace_knowledgebase, workspace_search
|
||||
|
||||
cfg = _config(tmp_path)
|
||||
extra = tmp_path / "notes"
|
||||
(extra / "nested").mkdir(parents=True)
|
||||
(extra / "top.txt").write_text("release notes\n", encoding="utf-8")
|
||||
(extra / "nested" / "deep.txt").write_text("hidden release notes\n", encoding="utf-8")
|
||||
cfg["knowledgebase"]["roots"] = [{"path": str(extra), "recursive": False}]
|
||||
|
||||
index_workspace_knowledgebase(cfg)
|
||||
result = workspace_search("release", config=cfg)
|
||||
|
||||
paths = {match["relative_path"] for match in result["matches"]}
|
||||
assert "notes/top.txt" in paths
|
||||
assert "notes/nested/deep.txt" not in paths
|
||||
|
||||
|
||||
class TestWorkspaceRetrieval:
|
||||
def test_index_workspace_builds_chunk_db_and_retrieves_ranked_chunks(self, tmp_path):
|
||||
from agent.workspace import index_workspace_knowledgebase, workspace_retrieve
|
||||
|
||||
cfg = _config(tmp_path)
|
||||
workspace = Path(cfg["workspace"]["path"])
|
||||
(workspace / "docs").mkdir(parents=True)
|
||||
(workspace / "docs" / "arch.md").write_text(
|
||||
"# Deployment\n\nThe deployment architecture uses blue green rollout and staged health checks.\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
(workspace / "notes").mkdir(parents=True)
|
||||
(workspace / "notes" / "random.txt").write_text("buy groceries\n", encoding="utf-8")
|
||||
|
||||
indexed = index_workspace_knowledgebase(cfg)
|
||||
assert indexed["success"] is True
|
||||
assert indexed["chunk_count"] >= 1
|
||||
assert Path(indexed["index_path"]).exists()
|
||||
|
||||
retrieved = workspace_retrieve("deployment architecture", config=cfg, limit=3)
|
||||
assert retrieved["success"] is True
|
||||
assert retrieved["count"] >= 1
|
||||
assert retrieved["results"][0]["relative_path"] == "docs/arch.md"
|
||||
assert "blue green" in retrieved["results"][0]["content"].lower()
|
||||
|
||||
def test_workspace_retrieve_reports_backend_metadata(self, tmp_path):
|
||||
from agent.workspace import index_workspace_knowledgebase, workspace_retrieve
|
||||
|
||||
cfg = _config(tmp_path)
|
||||
workspace = Path(cfg["workspace"]["path"])
|
||||
(workspace / "docs").mkdir(parents=True)
|
||||
(workspace / "docs" / "plan.md").write_text("blue green rollout plan\n", encoding="utf-8")
|
||||
|
||||
index_workspace_knowledgebase(cfg)
|
||||
retrieved = workspace_retrieve("blue green rollout", config=cfg, limit=2)
|
||||
|
||||
assert "dense_backend" in retrieved
|
||||
assert "rerank_backend" in retrieved
|
||||
|
||||
def test_workspace_context_for_turn_formats_sources_and_respects_gating(self, tmp_path):
|
||||
from agent.workspace import index_workspace_knowledgebase, workspace_context_for_turn
|
||||
|
||||
cfg = _config(tmp_path)
|
||||
cfg["knowledgebase"]["retrieval_mode"] = "always"
|
||||
workspace = Path(cfg["workspace"]["path"])
|
||||
(workspace / "docs").mkdir(parents=True)
|
||||
(workspace / "docs" / "plan.md").write_text(
|
||||
"Deployment plan includes canary analysis and rollback checkpoints.\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
index_workspace_knowledgebase(cfg)
|
||||
context = workspace_context_for_turn("summarize the deployment plan", config=cfg)
|
||||
assert "workspace context was retrieved for this turn only" in context.lower()
|
||||
assert "[source: relative/path]" in context.lower()
|
||||
assert "docs/plan.md" in context
|
||||
|
||||
cfg["knowledgebase"]["retrieval_mode"] = "gated"
|
||||
assert workspace_context_for_turn("thanks", config=cfg) == ""
|
||||
44
tests/hermes_cli/test_workspace_roots.py
Normal file
44
tests/hermes_cli/test_workspace_roots.py
Normal file
@@ -0,0 +1,44 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
def test_workspace_roots_add_defaults_to_non_recursive(tmp_path, monkeypatch):
|
||||
from hermes_cli.workspace import add_workspace_root
|
||||
|
||||
config = {
|
||||
"workspace": {"enabled": True, "path": str(tmp_path / "workspace")},
|
||||
"knowledgebase": {"enabled": True, "roots": []},
|
||||
}
|
||||
extra = tmp_path / "notes"
|
||||
extra.mkdir()
|
||||
|
||||
monkeypatch.setattr("hermes_cli.workspace.load_config", lambda: config)
|
||||
with patch("hermes_cli.workspace.save_config") as save_config:
|
||||
result = add_workspace_root(str(extra), recursive=False)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["root"]["recursive"] is False
|
||||
save_config.assert_called_once()
|
||||
|
||||
|
||||
def test_workspace_roots_remove_by_path(tmp_path, monkeypatch):
|
||||
from hermes_cli.workspace import remove_workspace_root
|
||||
|
||||
extra = tmp_path / "notes"
|
||||
config = {
|
||||
"workspace": {"enabled": True, "path": str(tmp_path / "workspace")},
|
||||
"knowledgebase": {
|
||||
"enabled": True,
|
||||
"roots": [{"path": str(extra), "recursive": False}],
|
||||
},
|
||||
}
|
||||
|
||||
monkeypatch.setattr("hermes_cli.workspace.load_config", lambda: config)
|
||||
with patch("hermes_cli.workspace.save_config") as save_config:
|
||||
result = remove_workspace_root(str(extra))
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["removed"]["path"] == str(extra)
|
||||
save_config.assert_called_once()
|
||||
22
tests/test_workspace_cli_command.py
Normal file
22
tests/test_workspace_cli_command.py
Normal file
@@ -0,0 +1,22 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
||||
class TestWorkspaceCLICommand:
|
||||
def _make_cli(self):
|
||||
from cli import HermesCLI
|
||||
|
||||
cli = HermesCLI.__new__(HermesCLI)
|
||||
cli.config = {"quick_commands": {}}
|
||||
cli.console = MagicMock()
|
||||
cli.agent = None
|
||||
cli.conversation_history = []
|
||||
return cli
|
||||
|
||||
def test_process_command_dispatches_workspace_handler(self):
|
||||
cli = self._make_cli()
|
||||
|
||||
with patch("hermes_cli.workspace.handle_workspace_slash") as handler:
|
||||
result = cli.process_command("/workspace status")
|
||||
|
||||
assert result is True
|
||||
handler.assert_called_once_with("/workspace status", console=cli.console)
|
||||
95
tests/tools/test_workspace_tool.py
Normal file
95
tests/tools/test_workspace_tool.py
Normal file
@@ -0,0 +1,95 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _config(tmp_path: Path) -> dict:
|
||||
return {
|
||||
"workspace": {
|
||||
"enabled": True,
|
||||
"path": str(tmp_path / "workspace"),
|
||||
"auto_create": True,
|
||||
"persist_gateway_uploads": "ask",
|
||||
},
|
||||
"knowledgebase": {
|
||||
"enabled": True,
|
||||
"path": str(tmp_path / "knowledgebase"),
|
||||
"roots": [],
|
||||
"retrieval_mode": "off",
|
||||
"auto_index": True,
|
||||
"watch_for_changes": False,
|
||||
"max_injected_chunks": 6,
|
||||
"max_injected_tokens": 3200,
|
||||
"dense_top_k": 40,
|
||||
"sparse_top_k": 40,
|
||||
"fused_top_k": 30,
|
||||
"final_top_k": 8,
|
||||
"min_fused_score": 0.0,
|
||||
"injection_format": "sourced_note",
|
||||
"chunking": {
|
||||
"default_tokens": 512,
|
||||
"overlap_tokens": 80,
|
||||
"code_strategy": "structural",
|
||||
"markdown_strategy": "headings",
|
||||
},
|
||||
"embeddings": {"provider": "local", "model": "google/embeddinggemma-300m", "dimensions": 768},
|
||||
"reranker": {"enabled": False, "provider": "local", "model": "bge-reranker-v2-m3"},
|
||||
"indexing": {
|
||||
"respect_gitignore": True,
|
||||
"respect_hermesignore": True,
|
||||
"include_hidden": False,
|
||||
"max_file_mb": 10,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class TestWorkspaceTool:
|
||||
def test_status_reports_workspace_roots(self, tmp_path, monkeypatch):
|
||||
from tools.workspace_tool import workspace_tool
|
||||
|
||||
monkeypatch.setattr("tools.workspace_tool.load_config", lambda: _config(tmp_path))
|
||||
|
||||
result = json.loads(workspace_tool(action="status"))
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["workspace_root"].endswith("workspace")
|
||||
assert result["knowledgebase_root"].endswith("knowledgebase")
|
||||
|
||||
def test_index_search_and_retrieve_round_trip(self, tmp_path, monkeypatch):
|
||||
from tools.workspace_tool import workspace_tool
|
||||
|
||||
cfg = _config(tmp_path)
|
||||
workspace = Path(cfg["workspace"]["path"])
|
||||
(workspace / "docs").mkdir(parents=True)
|
||||
(workspace / "docs" / "deploy.md").write_text("deployment checklist and rollback plan\n", encoding="utf-8")
|
||||
monkeypatch.setattr("tools.workspace_tool.load_config", lambda: cfg)
|
||||
|
||||
indexed = json.loads(workspace_tool(action="index"))
|
||||
assert indexed["success"] is True
|
||||
assert indexed["file_count"] == 1
|
||||
assert indexed["chunk_count"] >= 1
|
||||
|
||||
searched = json.loads(workspace_tool(action="search", query="deployment"))
|
||||
assert searched["success"] is True
|
||||
assert searched["count"] == 1
|
||||
assert searched["matches"][0]["relative_path"] == "docs/deploy.md"
|
||||
|
||||
retrieved = json.loads(workspace_tool(action="retrieve", query="rollback plan"))
|
||||
assert retrieved["success"] is True
|
||||
assert retrieved["count"] >= 1
|
||||
assert retrieved["results"][0]["relative_path"] == "docs/deploy.md"
|
||||
|
||||
def test_list_returns_relative_paths(self, tmp_path, monkeypatch):
|
||||
from tools.workspace_tool import workspace_tool
|
||||
|
||||
cfg = _config(tmp_path)
|
||||
workspace = Path(cfg["workspace"]["path"])
|
||||
(workspace / "notes").mkdir(parents=True)
|
||||
(workspace / "notes" / "todo.txt").write_text("ship it\n", encoding="utf-8")
|
||||
monkeypatch.setattr("tools.workspace_tool.load_config", lambda: cfg)
|
||||
|
||||
listed = json.loads(workspace_tool(action="list"))
|
||||
assert listed["success"] is True
|
||||
assert listed["entries"][0]["relative_path"] == "notes/todo.txt"
|
||||
123
tools/workspace_tool.py
Normal file
123
tools/workspace_tool.py
Normal file
@@ -0,0 +1,123 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Workspace tool — inspect and search the Hermes workspace."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
from agent.workspace import (
|
||||
index_workspace_knowledgebase,
|
||||
workspace_list,
|
||||
workspace_retrieve,
|
||||
workspace_search,
|
||||
workspace_status,
|
||||
)
|
||||
from hermes_cli.config import load_config
|
||||
from tools.registry import registry
|
||||
|
||||
|
||||
WORKSPACE_SCHEMA = {
|
||||
"name": "workspace",
|
||||
"description": "Manage the Hermes workspace under HERMES_HOME. Use this to inspect workspace status, rebuild the workspace manifest, list files, or search within workspace documents without relying on the terminal environment.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"action": {
|
||||
"type": "string",
|
||||
"enum": ["status", "index", "list", "search", "retrieve"],
|
||||
"description": "What to do: status shows roots and counts, index rebuilds the manifest and chunk index, list enumerates files, search searches text lines, retrieve returns ranked chunk-level retrieval results.",
|
||||
},
|
||||
"query": {
|
||||
"type": "string",
|
||||
"description": "Regex query to search for when action='search'.",
|
||||
},
|
||||
"path": {
|
||||
"type": "string",
|
||||
"description": "Optional subpath within the workspace to scope list/search operations.",
|
||||
},
|
||||
"file_glob": {
|
||||
"type": "string",
|
||||
"description": "Optional filename glob filter for search, e.g. '*.md'.",
|
||||
},
|
||||
"limit": {
|
||||
"type": "integer",
|
||||
"description": "Maximum number of entries or matches to return.",
|
||||
"default": 20,
|
||||
},
|
||||
"offset": {
|
||||
"type": "integer",
|
||||
"description": "Skip the first N entries or matches.",
|
||||
"default": 0,
|
||||
},
|
||||
"recursive": {
|
||||
"type": "boolean",
|
||||
"description": "When action='list', recurse through subdirectories (default true).",
|
||||
"default": True,
|
||||
},
|
||||
},
|
||||
"required": ["action"],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def workspace_tool(
|
||||
action: str,
|
||||
query: str = "",
|
||||
path: str = "",
|
||||
file_glob: str | None = None,
|
||||
limit: int = 20,
|
||||
offset: int = 0,
|
||||
recursive: bool = True,
|
||||
) -> str:
|
||||
try:
|
||||
config = load_config()
|
||||
if action == "status":
|
||||
result: dict[str, Any] = workspace_status(config)
|
||||
elif action == "index":
|
||||
result = index_workspace_knowledgebase(config)
|
||||
elif action == "list":
|
||||
result = workspace_list(
|
||||
config=config,
|
||||
relative_path=path,
|
||||
recursive=recursive,
|
||||
limit=limit,
|
||||
offset=offset,
|
||||
)
|
||||
elif action == "search":
|
||||
result = workspace_search(
|
||||
query=query,
|
||||
config=config,
|
||||
relative_path=path,
|
||||
file_glob=file_glob,
|
||||
limit=limit,
|
||||
offset=offset,
|
||||
)
|
||||
elif action == "retrieve":
|
||||
result = workspace_retrieve(
|
||||
query=query,
|
||||
config=config,
|
||||
limit=limit,
|
||||
)
|
||||
else:
|
||||
result = {"success": False, "error": f"Unknown action: {action}"}
|
||||
return json.dumps(result, ensure_ascii=False)
|
||||
except Exception as e: # pragma: no cover - defensive wrapper
|
||||
return json.dumps({"success": False, "error": str(e)}, ensure_ascii=False)
|
||||
|
||||
|
||||
registry.register(
|
||||
name="workspace",
|
||||
toolset="workspace",
|
||||
schema=WORKSPACE_SCHEMA,
|
||||
handler=lambda args, **kw: workspace_tool(
|
||||
action=args.get("action", ""),
|
||||
query=args.get("query", ""),
|
||||
path=args.get("path", ""),
|
||||
file_glob=args.get("file_glob"),
|
||||
limit=args.get("limit", 20),
|
||||
offset=args.get("offset", 0),
|
||||
recursive=args.get("recursive", True),
|
||||
),
|
||||
check_fn=lambda: True,
|
||||
)
|
||||
@@ -62,6 +62,8 @@ _HERMES_CORE_TOOLS = [
|
||||
"send_message",
|
||||
# Home Assistant smart home control (gated on HASS_TOKEN via check_fn)
|
||||
"ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service",
|
||||
# Workspace knowledge base
|
||||
"workspace",
|
||||
]
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user