Compare commits

...

1 Commits

Author SHA1 Message Date
Teknium
81d4467052 feat: add workspace foundation and RAG retrieval system
Port and modernize PR #1324 onto current main with full profile/HERMES_HOME awareness.

New files:
- agent/workspace.py: Core workspace engine — path resolution, manifest generation,
  structural chunking (markdown heading-aware, code symbol-aware), chunk indexing
  into SQLite, hybrid retrieval (FTS5 sparse + dense embeddings via RRF), optional
  reranking (local cross-encoder, Cohere, Voyage, heuristic fallback), workspace
  roots management, turn-scoped context injection
- tools/workspace_tool.py: Model-facing workspace tool (status/index/list/search/retrieve)
- hermes_cli/workspace.py: CLI subcommands and /workspace slash command handler

Integration points:
- config.py: workspace and knowledgebase sections in DEFAULT_CONFIG, workspace/
  knowledgebase dirs in ensure_hermes_home(), config version bump to 13
- toolsets.py: workspace tool added to _HERMES_CORE_TOOLS
- model_tools.py: workspace_tool added to _discover_tools()
- commands.py: /workspace CommandDef with subcommands
- cli.py: /workspace slash command dispatch
- run_agent.py: turn-scoped workspace RAG context injection (cache-safe —
  appended to current-turn user message only, never touches system prompt)
- hermes_cli/main.py: hermes workspace subcommand tree
  (status/index/list/search/retrieve/roots)
- hermes_cli/banner.py: workspace roots visibility in welcome banner
- pyproject.toml: workspace-rag optional dependency group

Profile-aware: all paths use get_hermes_home() from hermes_constants,
never hardcoded ~/.hermes. Each profile gets its own workspace/ and
knowledgebase/ directories.

Retrieval modes: off (default), gated (heuristic trigger), always.
Embedding: local SentenceTransformers when installed, hash fallback otherwise.
Dense search: sqlite-vec acceleration when installed, Python cosine fallback.

Tests: 18 new workspace-specific tests, all passing.
Original PR: #1324 by @teknium1
2026-04-06 12:55:18 -07:00
16 changed files with 2443 additions and 2 deletions

1455
agent/workspace.py Normal file

File diff suppressed because it is too large Load Diff

3
cli.py
View File

@@ -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":

View File

@@ -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

View File

@@ -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",

View File

@@ -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,
}
# =============================================================================

View File

@@ -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
View 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 ...]")

View File

@@ -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:

View File

@@ -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"]

View File

@@ -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)

View 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) == ""

View 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()

View 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)

View 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
View 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,
)

View File

@@ -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",
]