feat(viking): tool filtering, error diagnostics, and URI-prefix instructions

changelog/unreleased/2026-08-18-viking-allowed-uri-prefixes.md

@@ -19,4 +19,20 @@ single subtree such as `viking://resources/wiki/` without also allowing
 
 Empty list (the default) preserves unrestricted behavior for backward
 compatibility. `viking_search`/`viking_find` without a `target_uri`
-automatically scope to the first allowed prefix when a allowlist is set.
+automatically scope to the configured allowlist when one is set (see below).
+
+## Search scoping covers all allowed prefixes (2026-08-19)
+
+When a `target_uri` is omitted, `viking_search`/`viking_find` previously
+scoped to only the **first** allowed prefix, silently dropping results from
+the other allowed trees. `target_uri` now accepts a list, and both tools pass
+**every** allowed prefix to the SDK — the server searches each tree and the
+result set covers the full allowlist. Explicit `str` targets are still
+validated against the allowlist as before; explicit `list` targets are
+validated element-wise.
+
+Alongside this, `get_instructions()` now renders a dynamic **Allowed URI
+Prefixes** section listing the exact prefixes when the allowlist is
+configured. The model sees the scoping boundary up front, so it can pass the
+most specific `target_uri` directly and skip discovery probing (`viking_ls`),
+which also avoids the slower whole-allowlist search.

changelog/unreleased/2026-08-19-viking-tool-error-diagnostics.md

@@ -0,0 +1,18 @@
+# Viking tool errors now include the exception type
+
+All 15 `viking_*` tools previously rendered failures as
+`viking_search error: {e}`, relying on `str(e)` for the diagnostic text.
+Some exceptions carry an empty message — notably `httpx.ReadTimeout('')`
+when a slow knowledge-graph search exceeds the configured timeout — which
+produced a useless `viking_search error:` with no trailing context and no
+indication of what went wrong.
+
+Tool error returns now follow `viking_search error ({ExcType}): {e}`, so
+an empty-message timeout renders as `viking_search error (ReadTimeout):`
+and the failure class is always identifiable even when the message is
+blank. This applies uniformly to all 15 tools (search, find, recall,
+grep, glob, ls, read, expand, write, edit, mkdir, add_resource, forget,
+link, set_tags).
+
+Also adds a regression test asserting that empty-message exceptions still
+surface their exception type in the tool return value.

changelog/unreleased/2026-08-19-viking-tool-filtering.md

@@ -0,0 +1,26 @@
+# Viking tool filtering via enabled_tools / disabled_tools
+
+`VikingCapabilityConfig` gains two mutually-exclusive tool-filter fields,
+mirroring `StdioMCPServerConfig`:
+
+- `enabled_tools` — whitelist: only these `viking_*` tools are exposed.
+- `disabled_tools` — blacklist: these tools are excluded from the exposed set.
+
+The filter applies in `build_tools()` after mode-based assembly, so it works
+for every mode (`retrieve`/`write`/`graph`/`all`) and propagates to both the
+toolset and the OpenCode `/experimental/tool` listing via `get_tools()`.
+
+Motivation: the knowledge-graph semantic search backend
+(`viking_search`/`viking_find`) can be slow on large resource trees (40-60s+
+per query), exceeding typical client timeouts. Operators can now disable
+just those tools while keeping the deterministic ones:
+
+```yaml
+capabilities:
+  - type: viking
+    mode: retrieve
+    disabled_tools: ["viking_search", "viking_find"]
+```
+
+Specifying both `enabled_tools` and `disabled_tools` raises a validation
+error (mutually exclusive), matching the MCP server config pattern.

src/wolfharness/capabilities/viking/__init__.py

@@ -217,6 +217,14 @@ class VikingCapability(AbstractCapability[Any]):
     """When True (default), profile injection runs only on the first turn
     of a session. When False, injection runs on every ``before_model_request``
     call (not recommended — expensive and static)."""
+    enabled_tools: list[str] | None = None
+    """If set, only these tools are exposed (whitelist). Mutually exclusive
+    with ``disabled_tools``."""
+    disabled_tools: list[str] | None = None
+    """Tools to exclude from the exposed set (blacklist). Mutually exclusive
+    with ``enabled_tools``. For example, disable a slow semantic-search
+    backend while keeping deterministic tools:
+    ``["viking_search", "viking_find"]``."""
     compaction_enabled: bool = False
     """When True, archive old conversation messages to Viking before
     context overflow. Disabled by default."""
@@ -545,12 +553,21 @@ async def for_run(self, ctx: RunContext[Any]) -> VikingCapability:
     def get_instructions(self) -> str | None:
         """Return the Viking workflow instructions.
 
+        When ``allowed_uri_prefixes`` is configured, appends a dynamic
+        block listing the allowed prefixes so the model can pass a
+        ``target_uri`` and skip discovery probing.
+
         Returns:
             The instruction string from ``instructions.py``.
         """
-        from wolfharness.capabilities.viking.instructions import _VIKING_INSTRUCTIONS
+        from wolfharness.capabilities.viking.instructions import (
+            _VIKING_INSTRUCTIONS,
+            format_allowed_prefixes_block,
+        )
 
-        return _VIKING_INSTRUCTIONS
+        if not self.allowed_uri_prefixes:
+            return _VIKING_INSTRUCTIONS
+        return _VIKING_INSTRUCTIONS + format_allowed_prefixes_block(self.allowed_uri_prefixes)
 
     def get_toolset(self) -> AgentToolset[Any] | None:
         """Build a ``FunctionToolset`` from tools filtered by ``self.mode``.

src/wolfharness/capabilities/viking/instructions.py

@@ -11,6 +11,39 @@
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+
+def format_allowed_prefixes_block(prefixes: Sequence[str]) -> str:
+    """Build the allowed-prefix instruction block.
+
+    Appended to the base instructions when ``allowed_uri_prefixes`` is
+    configured. Rendered dynamically at ``get_instructions()`` time so the
+    model sees the exact prefixes this session may access — it can then pass
+    the most specific prefix as ``target_uri`` and skip discovery probing.
+
+    Args:
+        prefixes: The configured allowed URI prefixes.
+
+    Returns:
+        A markdown block listing the prefixes and the search guidance.
+    """
+    allowed = "\n".join(f"  - `{p}`" for p in prefixes)
+    return (
+        "\n### Allowed URI Prefixes (this session)\n"
+        "Your Viking access is restricted to these prefixes:\n"
+        f"{allowed}\n"
+        "When calling `viking_search` / `viking_find`, ALWAYS pass the most "
+        "specific matching prefix as `target_uri` — it scopes the search and "
+        "is faster. Omitting it searches all allowed prefixes, which is slower. "
+        "Use `viking_ls` on these prefixes to see what is in scope. Do not "
+        "probe outside them — you will get access errors."
+    )
+
 
 _VIKING_INSTRUCTIONS = """\
 ## Viking Knowledge Graph Tools

src/wolfharness/capabilities/viking/tools.py

@@ -93,7 +93,7 @@ async def viking_search(
             limit: int = 10,
             min_score: float = 0.35,
             level: list[int] | None = None,
-            target_uri: str = "",
+            target_uri: str | list[str] = "",
         ) -> ToolReturn:
             """Search the Viking knowledge graph semantically.
 
@@ -105,7 +105,8 @@ async def viking_search(
                 limit: Maximum number of results to return.
                 min_score: Minimum relevance score (0.0 to 1.0).
                 level: Filter by content level (e.g. [0, 1, 2] for L0-L2).
-                target_uri: Restrict search to a specific URI subtree.
+                target_uri: Restrict search to specific URI subtrees — a
+                    single ``viking://`` URI or a list of them.
 
             Returns:
                 Formatted search results grouped by context type.
@@ -115,12 +116,22 @@ async def viking_search(
                 sid = _get_session_id(ctx)
                 sdk_filter: dict[str, Any] | None = {"level": level} if level else None
                 if cap.allowed_uri_prefixes:
-                    if target_uri:
-                        err = cap._check_uri_allowed(target_uri, tool_name="viking_search")
-                        if err:
-                            return ToolReturn(return_value=err)
+                    if isinstance(target_uri, str):
+                        if target_uri:
+                            err = cap._check_uri_allowed(target_uri, tool_name="viking_search")
+                            if err:
+                                return ToolReturn(return_value=err)
                     else:
-                        target_uri = cap.allowed_uri_prefixes[0]
+                        for u in target_uri:
+                            err = cap._check_uri_allowed(u, tool_name="viking_search")
+                            if err:
+                                return ToolReturn(return_value=err)
+                    if not target_uri:
+                        # SDK target_uri accepts a list — the server searches
+                        # every allowed prefix. The old default used only the
+                        # first prefix, silently dropping the other allowed
+                        # trees.
+                        target_uri = cap.allowed_uri_prefixes
                 result = await client.search(
                     query,
                     target_uri=target_uri,
@@ -131,15 +142,15 @@ async def viking_search(
                 )
                 return ToolReturn(return_value=format_search_results(result))
             except Exception as e:
-                return ToolReturn(return_value=f"viking_search error: {e}")
+                return ToolReturn(return_value=f"viking_search error ({type(e).__name__}): {e}")
 
         async def viking_find(
             ctx: RunContext[Any],
             query: str,
             limit: int = 10,
             min_score: float = 0.35,
             level: list[int] | None = None,
-            target_uri: str = "",
+            target_uri: str | list[str] = "",
         ) -> ToolReturn:
             """Find content in Viking, deduplicating results.
 
@@ -151,7 +162,8 @@ async def viking_find(
                 limit: Maximum number of results to return.
                 min_score: Minimum relevance score (0.0 to 1.0).
                 level: Filter by content level (e.g. [0, 1, 2] for L0-L2).
-                target_uri: Restrict search to a specific URI subtree.
+                target_uri: Restrict search to specific URI subtrees — a
+                    single ``viking://`` URI or a list of them.
 
             Returns:
                 Formatted search results grouped by context type.
@@ -160,12 +172,20 @@ async def viking_find(
                 client = await cap._ensure_client()
                 sdk_filter: dict[str, Any] | None = {"level": level} if level else None
                 if cap.allowed_uri_prefixes:
-                    if target_uri:
-                        err = cap._check_uri_allowed(target_uri, tool_name="viking_find")
-                        if err:
-                            return ToolReturn(return_value=err)
+                    if isinstance(target_uri, str):
+                        if target_uri:
+                            err = cap._check_uri_allowed(target_uri, tool_name="viking_find")
+                            if err:
+                                return ToolReturn(return_value=err)
                     else:
-                        target_uri = cap.allowed_uri_prefixes[0]
+                        for u in target_uri:
+                            err = cap._check_uri_allowed(u, tool_name="viking_find")
+                            if err:
+                                return ToolReturn(return_value=err)
+                    if not target_uri:
+                        # See viking_search: SDK target_uri accepts a list so we
+                        # search every allowed prefix, not just the first.
+                        target_uri = cap.allowed_uri_prefixes
                 result = await client.find(
                     query,
                     target_uri=target_uri,
@@ -175,7 +195,7 @@ async def viking_find(
                 )
                 return ToolReturn(return_value=format_search_results(result))
             except Exception as e:
-                return ToolReturn(return_value=f"viking_find error: {e}")
+                return ToolReturn(return_value=f"viking_find error ({type(e).__name__}): {e}")
 
         async def viking_recall(
             ctx: RunContext[Any],
@@ -243,7 +263,7 @@ async def viking_recall(
                 merged = "\n\n".join(sections)
                 return ToolReturn(return_value=truncate_text(merged, max_chars))
             except Exception as e:
-                return ToolReturn(return_value=f"viking_recall error: {e}")
+                return ToolReturn(return_value=f"viking_recall error ({type(e).__name__}): {e}")
 
         async def viking_grep(
             ctx: RunContext[Any],
@@ -300,7 +320,7 @@ async def _grep_one(p: str) -> list[dict[str, Any]]:
 
                 return ToolReturn(return_value=format_grep_results(all_matches, patterns))
             except Exception as e:
-                return ToolReturn(return_value=f"viking_grep error: {e}")
+                return ToolReturn(return_value=f"viking_grep error ({type(e).__name__}): {e}")
 
         async def viking_glob(
             ctx: RunContext[Any],
@@ -333,7 +353,7 @@ async def viking_glob(
                     uris = result if isinstance(result, list) else []
                 return ToolReturn(return_value=format_glob_results([str(u) for u in uris], pattern))
             except Exception as e:
-                return ToolReturn(return_value=f"viking_glob error: {e}")
+                return ToolReturn(return_value=f"viking_glob error ({type(e).__name__}): {e}")
 
         async def viking_ls(
             ctx: RunContext[Any],
@@ -416,7 +436,7 @@ async def _safe_abstract(entry_uri: str) -> str:
 
                 return ToolReturn(return_value=format_ls_entries(entry_list))
             except Exception as e:
-                return ToolReturn(return_value=f"viking_ls error: {e}")
+                return ToolReturn(return_value=f"viking_ls error ({type(e).__name__}): {e}")
 
         async def viking_read(
             ctx: RunContext[Any],
@@ -510,7 +530,7 @@ async def viking_read(
                     )
                 return ToolReturn(return_value="\n\n".join(sections))
             except Exception as e:
-                return ToolReturn(return_value=f"viking_read error: {e}")
+                return ToolReturn(return_value=f"viking_read error ({type(e).__name__}): {e}")
 
         async def viking_expand(
             ctx: RunContext[Any],
@@ -539,7 +559,7 @@ async def viking_expand(
                     return_value=str(content) if content else "No content found at URI."
                 )
             except Exception as e:
-                return ToolReturn(return_value=f"viking_expand error: {e}")
+                return ToolReturn(return_value=f"viking_expand error ({type(e).__name__}): {e}")
 
         retrieve_tools: list[Callable[..., Awaitable[ToolReturn]]] = [
             viking_search,
@@ -618,7 +638,7 @@ async def viking_write(
                     return_value=f"Wrote {len(content)} chars to {uri} (mode={mode})."
                 )
             except Exception as e:
-                return ToolReturn(return_value=f"viking_write error: {e}")
+                return ToolReturn(return_value=f"viking_write error ({type(e).__name__}): {e}")
 
         async def viking_edit(
             ctx: RunContext[Any],
@@ -669,7 +689,7 @@ async def viking_edit(
                 await client.write(uri, modified, mode="replace")
                 return ToolReturn(return_value=f"Replaced {count} occurrence(s) in {uri}.")
             except Exception as e:
-                return ToolReturn(return_value=f"viking_edit error: {e}")
+                return ToolReturn(return_value=f"viking_edit error ({type(e).__name__}): {e}")
 
         async def viking_mkdir(
             ctx: RunContext[Any],
@@ -692,7 +712,7 @@ async def viking_mkdir(
                 await client.mkdir(uri, description=description)
                 return ToolReturn(return_value=f"Created directory {uri}.")
             except Exception as e:
-                return ToolReturn(return_value=f"viking_mkdir error: {e}")
+                return ToolReturn(return_value=f"viking_mkdir error ({type(e).__name__}): {e}")
 
         async def viking_add_resource(
             ctx: RunContext[Any],
@@ -736,7 +756,9 @@ async def viking_add_resource(
                 )
                 return ToolReturn(return_value=f"Added resource {path} to Viking. Result: {result}")
             except Exception as e:
-                return ToolReturn(return_value=f"viking_add_resource error: {e}")
+                return ToolReturn(
+                    return_value=f"viking_add_resource error ({type(e).__name__}): {e}"
+                )
 
         async def viking_forget(
             ctx: RunContext[Any],
@@ -759,7 +781,7 @@ async def viking_forget(
                 await client.rm(uri, recursive=recursive)
                 return ToolReturn(return_value=f"Removed {uri}.")
             except Exception as e:
-                return ToolReturn(return_value=f"viking_forget error: {e}")
+                return ToolReturn(return_value=f"viking_forget error ({type(e).__name__}): {e}")
 
         write_tools: list[Callable[..., Awaitable[ToolReturn]]] = [
             viking_write,
@@ -809,7 +831,7 @@ async def viking_link(
                     return_value=f"Linked {from_uri} -> {', '.join(targets)} (reason: {reason!r})."
                 )
             except Exception as e:
-                return ToolReturn(return_value=f"viking_link error: {e}")
+                return ToolReturn(return_value=f"viking_link error ({type(e).__name__}): {e}")
 
         async def viking_set_tags(
             ctx: RunContext[Any],
@@ -834,11 +856,18 @@ async def viking_set_tags(
                 await client.set_tags(uri, tags, mode="replace", recursive=recursive)
                 return ToolReturn(return_value=f"Set {len(tags)} tag(s) on {uri}.")
             except Exception as e:
-                return ToolReturn(return_value=f"viking_set_tags error: {e}")
+                return ToolReturn(return_value=f"viking_set_tags error ({type(e).__name__}): {e}")
 
         graph_tools: list[Callable[..., Awaitable[ToolReturn]]] = [viking_set_tags]
         if cap.enable_link:
             graph_tools.append(viking_link)
         tools.extend(graph_tools)
 
+    if cap.enabled_tools is not None:
+        names = {getattr(fn, "__name__", "") for fn in tools}
+        allowed = {n for n in cap.enabled_tools if n in names}
+        tools = [fn for fn in tools if getattr(fn, "__name__", "") in allowed]
+    elif cap.disabled_tools is not None:
+        tools = [fn for fn in tools if getattr(fn, "__name__", "") not in set(cap.disabled_tools)]
+
     return tools