Add provider-aware search support to hybrid orchestration agents (#558)

Previously `core/config.py` defined `DEFAULT_CONFIG_DIR = Path.home() /
".openjarvis"` as a by-value module constant imported into ~45 modules, and 34
modules hardcoded `Path.home() / ".openjarvis"` directly. The installer honored
`OPENJARVIS_HOME` but the Python runtime ignored it, producing a split-brain
layout (some modules honored the override, the core config dir did not). Eval
dataset caches also scattered into `~/.cache/<benchmark>`.

This introduces a single env-aware resolver in `openjarvis/core/paths.py` and
routes every state/config/cache path through it. OpenJarvis now keeps ALL of
its state under ONE root, resolved in priority order:

  1. $OPENJARVIS_HOME
  2. $XDG_DATA_HOME/openjarvis   (single nested dir, when XDG_DATA_HOME is set)
  3. ~/.openjarvis               (default — unchanged, so existing installs are
                                  untouched and no data migration is required)

Implementation:
- New `core/paths.py`: get_config_dir / get_config_path / get_data_dir /
  get_cache_dir, with a source-tree rejection guard (fails loudly per
  REVIEW.md if the root resolves inside the repo).
- `core/config.py`: DEFAULT_CONFIG_DIR / DEFAULT_CONFIG_PATH are now resolved
  via the env-aware resolver at import (real attributes, so existing
  monkeypatch.setattr-based tests keep working). All dataclass field defaults
  that pointed at ~/.openjarvis converted to default_factory so they honor the
  override at instantiation.
- Routed all 34 hardcoders plus several string-literal escapees the original
  audit missed: prompt_loader / description_loader (were OPENJARVIS_HOME-only,
  no XDG), swebench_harness cache, tools/{memory,skill,user_profile}_manage
  defaults, server trace.db fallbacks, doctor_cmd hints.
- spec_search storage/paths now delegates to the unified resolver (gains XDG);
  its ConfigurationError is aliased to the core one.
- Eval dataset caches moved from ~/.cache/<name> to <root>/cache/<name>
  (~/.cache/huggingface left alone — it is HF's own cache).
- Docs + installer comment + `jarvis config path` to show resolved dirs.

Read-only macOS connectors and OS service files (LaunchAgents/systemd) are
intentionally left untouched.

Fixes #462

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.github/clone-stats/badge.json

@@ -1,7 +1,7 @@
 {
   "schemaVersion": 1,
   "label": "Git Clones",
-  "message": "117,047",
+  "message": "122,456",
   "color": "green",
   "namedLogo": "git"
 }

.github/clone-stats/clone-data.json

@@ -1,6 +1,6 @@
 {
-  "total_clones": 117047,
-  "last_updated": "2026-06-14T07:38:47Z",
+  "total_clones": 122456,
+  "last_updated": "2026-06-18T07:44:28Z",
   "daily": {
     "2026-03-27": 2189,
     "2026-03-28": 1874,
@@ -80,6 +80,10 @@
     "2026-06-10": 1310,
     "2026-06-11": 2564,
     "2026-06-12": 1313,
-    "2026-06-13": 2804
+    "2026-06-13": 2804,
+    "2026-06-14": 1543,
+    "2026-06-15": 1379,
+    "2026-06-16": 1317,
+    "2026-06-17": 1170
   }
 }

docs/getting-started/configuration.md

@@ -17,6 +17,46 @@ The configuration file lives at:
 
 OpenJarvis creates the `~/.openjarvis/` directory and populates it with a default config when you run `jarvis init`.
 
+## Relocating the OpenJarvis directory
+
+OpenJarvis keeps **all** of its state — config, databases, caches, logs,
+credentials, skills, recipes, connectors — under a **single root** so it never
+clutters your home directory beyond one folder. By default that root is
+`~/.openjarvis`, but you can move it.
+
+The root is resolved in priority order:
+
+1. **`$OPENJARVIS_HOME`** — explicit override. Honored by both the installer
+   and the Python runtime.
+2. **`$XDG_DATA_HOME/openjarvis`** — used when `$XDG_DATA_HOME` is set (a single
+   `openjarvis` directory nested under it, per the XDG Base Directory spec).
+3. **`~/.openjarvis`** — the default. With no environment variables set, the
+   resolved path is exactly this, so existing installs are untouched.
+
+```bash
+# Relocate the whole install + runtime tree at install time:
+OPENJARVIS_HOME=~/apps/openjarvis curl -fsSL https://open-jarvis.github.io/OpenJarvis/install.sh | bash
+
+# Or for a single run / your shell profile:
+export OPENJARVIS_HOME=~/apps/openjarvis
+```
+
+Confirm where your data lives with:
+
+```bash
+jarvis config path
+```
+
+!!! note "Migration"
+    Because the default is unchanged, **no data migration is required** for
+    existing installs. If you set `OPENJARVIS_HOME` (or `XDG_DATA_HOME`) on a
+    machine that already has data in `~/.openjarvis`, OpenJarvis will look in
+    the new location and not see your old data — move it yourself if you want
+    to keep it: `mv ~/.openjarvis "$OPENJARVIS_HOME"`.
+
+`$OPENJARVIS_CONFIG` still points at an explicit `config.toml` file
+independently of the root, if you need to override just the config file path.
+
 ## Generating Configuration
 
 ### First-Time Setup

scripts/install/install.sh

@@ -210,6 +210,14 @@ if ! command -v python3 >/dev/null 2>&1; then
 fi
 
 # ---- env ----
+# OpenJarvis keeps ALL of its state (install tree + runtime data, configs,
+# databases, caches, logs) under a single root so it never clutters $HOME
+# beyond one directory. Relocate it by exporting OPENJARVIS_HOME before
+# running the installer, e.g.:
+#     OPENJARVIS_HOME=~/apps/openjarvis curl ... | bash
+# The Python runtime honors the same override (and, when OPENJARVIS_HOME is
+# unset, $XDG_DATA_HOME/openjarvis if XDG_DATA_HOME is set). With nothing set
+# the root is ~/.openjarvis, so existing installs are untouched.
 OPENJARVIS_HOME="${OPENJARVIS_HOME:-$HOME/.openjarvis}"
 OPENJARVIS_REPO_URL="${OPENJARVIS_REPO_URL:-https://github.com/open-jarvis/OpenJarvis.git}"
 SRC_DIR="$OPENJARVIS_HOME/src"

src/openjarvis/agents/claude_code.py

@@ -19,6 +19,7 @@ from typing import Any, List, Optional
 
 from openjarvis.agents._stubs import AgentContext, AgentResult, BaseAgent
 from openjarvis.core.events import EventBus
+from openjarvis.core.paths import get_config_dir
 from openjarvis.core.registry import AgentRegistry
 from openjarvis.core.types import ToolResult
 from openjarvis.engine._stubs import InferenceEngine
@@ -103,7 +104,7 @@ class ClaudeCodeAgent(BaseAgent):
                 "Install it from https://nodejs.org/ or via your package manager."
             )
 
-        dest = Path.home() / ".openjarvis" / "claude_code_runner"
+        dest = get_config_dir() / "claude_code_runner"
         dest.mkdir(parents=True, exist_ok=True)
 
         # Copy runner files if missing or outdated

src/openjarvis/agents/digest_store.py

@@ -9,6 +9,8 @@ from datetime import datetime
 from pathlib import Path
 from typing import Dict, List, Optional
 
+from openjarvis.core.paths import get_config_dir
+
 
 @dataclass
 class DigestArtifact:
@@ -30,7 +32,7 @@ class DigestStore:
 
     def __init__(self, db_path: str = "") -> None:
         if not db_path:
-            db_path = str(Path.home() / ".openjarvis" / "digest.db")
+            db_path = str(get_config_dir() / "digest.db")
         self._db_path = db_path
         self._conn = sqlite3.connect(db_path, check_same_thread=False)
         self._conn.execute("PRAGMA journal_mode=WAL")

src/openjarvis/agents/hybrid/_base.py

@@ -73,12 +73,19 @@ WEB_SEARCH_COST_PER_CALL = 0.01
 # $0.01/call number — kept as a separate constant so it can drift.
 OPENAI_WEB_SEARCH_COST_PER_CALL = 0.01
 
-# Gemini Google-Search grounding: billed at $35 per 1000 grounded
-# *requests* (2025-12 public list price for the Grounding-with-Google-Search
-# tool, charged once per request that uses the tool regardless of how many
-# internal queries it issues). We charge per grounded request, not per
-# `web_search_queries` entry.
-GEMINI_SEARCH_COST_PER_CALL = 0.035
+# Gemini 3 Google-Search grounding: billed at $14 per 1000 search queries.
+# `_call_gemini_agent` reports the model's `web_search_queries`, so this is
+# charged per query, not per outer generate_content request.
+GEMINI_SEARCH_COST_PER_CALL = 0.014
+
+# Tavily Search, advanced depth: 2 API credits per search request at $0.008
+# per credit on the public pay-as-you-go plan. WebSearchTool captures actual
+# credits when Tavily returns usage metadata; this is the fallback estimate.
+TAVILY_SEARCH_COST_PER_CREDIT = 0.008
+TAVILY_ADVANCED_SEARCH_CREDITS = 2
+TAVILY_SEARCH_COST_PER_CALL = (
+    TAVILY_SEARCH_COST_PER_CREDIT * TAVILY_ADVANCED_SEARCH_CREDITS
+)
 
 ANTHROPIC_WEB_SEARCH_TOOL = {
     "type": "web_search_20250305",
@@ -101,6 +108,40 @@ def build_web_search_tool(max_uses: int = 8) -> Dict[str, Any]:
     }
 
 
+def tavily_search_context(
+    query: str,
+    *,
+    max_results: int = 5,
+) -> Dict[str, Any]:
+    """Run OpenJarvis WebSearchTool and return accounting-friendly metadata."""
+    from openjarvis.tools.web_search import WebSearchTool
+
+    tool = WebSearchTool(max_results=max_results)
+    res = tool.execute(query=query, max_results=max_results)
+    meta = dict(res.metadata or {})
+    engine = str(meta.get("engine") or "unknown")
+    credits = 0
+    cost_usd = 0.0
+    if engine == "tavily":
+        try:
+            credits = int(meta.get("credits") or TAVILY_ADVANCED_SEARCH_CREDITS)
+        except (TypeError, ValueError):
+            credits = TAVILY_ADVANCED_SEARCH_CREDITS
+        cost_usd = credits * TAVILY_SEARCH_COST_PER_CREDIT
+    text = res.content or ""
+    if not res.success and not text:
+        text = "(no search results)"
+    return {
+        "text": text,
+        "success": bool(res.success),
+        "engine": engine,
+        "credits": credits,
+        "cost_usd": cost_usd,
+        "n_searches": 1 if (query or "").strip() else 0,
+        "error": None if res.success else text,
+    }
+
+
 def web_search_cfg(method_cfg: Optional[Dict[str, Any]]) -> Tuple[bool, int]:
     """Parse ``method_cfg.web_search = { enabled, max_uses }``.
 
@@ -1322,6 +1363,9 @@ __all__ = [
     "LocalCloudAgent",
     "NO_TEMP_PREFIXES",
     "OPENAI_WEB_SEARCH_COST_PER_CALL",
+    "TAVILY_ADVANCED_SEARCH_CREDITS",
+    "TAVILY_SEARCH_COST_PER_CALL",
+    "TAVILY_SEARCH_COST_PER_CREDIT",
     "WEB_SEARCH_COST_PER_CALL",
     "_bump_cloud_calls",
     "_bump_local_calls",
@@ -1329,5 +1373,6 @@ __all__ = [
     "estimate_cost",
     "is_gpt5_family",
     "supports_temperature",
+    "tavily_search_context",
     "web_search_cfg",
 ]

src/openjarvis/agents/hybrid/_prices.py

@@ -15,13 +15,16 @@ PRICES: dict[str, tuple[float, float]] = {
     "claude-sonnet-4-6":           (3.00, 15.0),
     "claude-haiku-4-5":            (1.00, 5.00),
     "claude-haiku-4-5-20251001":   (1.00, 5.00),
+    "gpt-5.5":                     (5.00, 30.0),
     "gpt-5":                       (1.25, 10.0),
     "gpt-5-mini":                  (0.25, 2.00),
     "gpt-5-mini-2025-08-07":       (0.25, 2.00),
     "gpt-4o":                      (0.15, 0.60),
-    # Gemini Developer API prices (USD per 1M tokens), 2025-12 list price.
-    # 2.5 Pro uses tiered pricing (>200K context = $2.50/$15); we charge the
-    # low-context tier since GAIA / SWE-bench prompts stay well under 200K.
+    # Gemini Developer API prices (USD per 1M tokens). Pro models use tiered
+    # pricing above 200K prompt tokens; GAIA prompts stay under that tier, so
+    # charge the low-context standard rate.
+    "gemini-3.1-pro-preview":       (2.00, 12.0),
+    "gemini-3.1-pro-preview-customtools": (2.00, 12.0),
     "gemini-2.5-pro":              (1.25, 10.0),
     "gemini-2.5-flash":            (0.30, 2.50),
     "gemini-2.5-flash-lite":       (0.10, 0.40),
@@ -61,7 +64,11 @@ def is_reasoning_model(model: str) -> bool:
     before emitting visible answer text. At max_tokens=4096 these silently
     truncate with empty answers on GAIA (26/100 GPT-5, 18/100 Gemini Pro)."""
     m = (model or "").lower()
-    return is_gpt5_family(model) or "gemini-2.5-pro" in m
+    return (
+        is_gpt5_family(model)
+        or "gemini-2.5-pro" in m
+        or "gemini-3.1-pro" in m
+    )
 
 
 def default_max_output_tokens(model: str) -> int:

src/openjarvis/agents/hybrid/advisors.py

@@ -34,6 +34,7 @@ from openjarvis.agents.hybrid._base import (
     WEB_SEARCH_COST_PER_CALL,
     LocalCloudAgent,
     build_web_search_tool,
+    tavily_search_context,
     web_search_cfg,
 )
 from openjarvis.agents.hybrid.mini_swe_agent import (
@@ -135,7 +136,12 @@ class AdvisorsAgent(LocalCloudAgent):
         advisor_temperature = float(cfg.get("advisor_temperature", 0.2))
 
         ws_enabled, ws_max_uses = web_search_cfg(cfg)
-        if ws_enabled and self._cloud_endpoint not in _SEARCH_CAPABLE_ENDPOINTS:
+        search_backend = str(cfg.get("search_backend", "provider")).lower()
+        if (
+            ws_enabled
+            and search_backend != "tavily"
+            and self._cloud_endpoint not in _SEARCH_CAPABLE_ENDPOINTS
+        ):
             raise ValueError(
                 f"web_search.enabled=true but cloud_endpoint={self._cloud_endpoint!r}; "
                 "server-side web_search is wired for anthropic / openai / gemini "
@@ -146,19 +152,23 @@ class AdvisorsAgent(LocalCloudAgent):
         use_ws = ws_enabled
         gaia_max_turns = int(cfg.get("gaia_max_turns", 8))
         n_searches_total = 0
+        search_cost_total = 0.0
 
         # 1. Initial executor pass — advisor (Qwen) doesn't get tools;
         # only the cloud executor passes do. With web_search on, dispatch
         # to the search-capable agent loop for the configured provider.
         if use_ws:
-            initial_resp, e1_in, e1_out, n_s1, e1_turns = self._executor_search(
+            (initial_resp, e1_in, e1_out, n_s1, e1_turns,
+             e1_search_cost) = self._executor_search(
                 user=f"Question:\n{question}",
                 system=EXECUTOR_INITIAL_SYS,
                 max_tokens=executor_max_tokens,
                 ws_max_uses=ws_max_uses,
                 max_turns=gaia_max_turns,
+                query=question,
             )
             n_searches_total += n_s1
+            search_cost_total += e1_search_cost
         else:
             initial_resp, e1_in, e1_out = self._call_cloud(
                 user=f"Question:\n{question}",
@@ -196,14 +206,17 @@ class AdvisorsAgent(LocalCloudAgent):
             f"answer-format rules."
         )
         if use_ws:
-            final_answer, e2_in, e2_out, n_s2, e2_turns = self._executor_search(
+            (final_answer, e2_in, e2_out, n_s2, e2_turns,
+             e2_search_cost) = self._executor_search(
                 user=final_user,
                 system=EXECUTOR_FINAL_SYS,
                 max_tokens=executor_max_tokens,
                 ws_max_uses=ws_max_uses,
                 max_turns=gaia_max_turns,
+                query=question,
             )
             n_searches_total += n_s2
+            search_cost_total += e2_search_cost
         else:
             final_answer, e2_in, e2_out = self._call_cloud(
                 user=final_user,
@@ -216,7 +229,10 @@ class AdvisorsAgent(LocalCloudAgent):
         tokens_local = adv_in + adv_out
         tokens_cloud = e1_in + e1_out + e2_in + e2_out
         cost = self.cost_usd(self._cloud_model, e1_in + e2_in, e1_out + e2_out)
-        cost += n_searches_total * _search_cost_per_call(self._cloud_endpoint)
+        if search_backend == "tavily":
+            cost += search_cost_total
+        else:
+            cost += n_searches_total * _search_cost_per_call(self._cloud_endpoint)
 
         meta: Dict[str, Any] = {
             "tokens_local": tokens_local,
@@ -233,7 +249,9 @@ class AdvisorsAgent(LocalCloudAgent):
                 "initial_response": initial_resp,
                 "advisor_feedback": advisor_text,
                 "web_search_enabled": use_ws,
+                "search_backend": search_backend,
                 "n_web_searches": n_searches_total,
+                "search_cost_usd": search_cost_total,
                 "note": "inference-only advisor (untrained); lower bound on the technique.",
             },
         }
@@ -251,16 +269,40 @@ class AdvisorsAgent(LocalCloudAgent):
         max_tokens: int,
         ws_max_uses: int,
         max_turns: int,
-    ) -> Tuple[str, int, int, int, int]:
+        query: Optional[str] = None,
+    ) -> Tuple[str, int, int, int, int, float]:
         """Run a search-capable executor pass for the configured cloud.
 
         Dispatches by ``self._cloud_endpoint`` to the matching ``_base``
-        agent loop. Returns the shared 5-tuple ``(text, p_tok, c_tok,
-        n_searches, turns)``. The endpoint is assumed already validated
-        against ``_SEARCH_CAPABLE_ENDPOINTS`` by the caller.
+        agent loop, or through Tavily when ``method_cfg.search_backend`` is
+        ``"tavily"``. Returns ``(text, p_tok, c_tok, n_searches, turns,
+        search_cost_usd)``.
         """
+        if str(self._cfg.get("search_backend", "provider")).lower() == "tavily":
+            res = tavily_search_context(
+                query or user,
+                max_results=int(self._cfg.get("tavily_max_results", 5)),
+            )
+            grounded_user = (
+                f"Web search results:\n{res['text']}\n\n"
+                f"Using the search results above, answer this request:\n{user}"
+            )
+            text, p, c = self._call_cloud(
+                user=grounded_user,
+                system=system,
+                max_tokens=max_tokens,
+                temperature=0.0,
+            )
+            return (
+                text,
+                p,
+                c,
+                int(res["n_searches"]),
+                1,
+                float(res["cost_usd"]),
+            )
         if self._cloud_endpoint == "anthropic":
-            return self._call_anthropic_agent(
+            text, p, c, n_searches, turns = self._call_anthropic_agent(
                 self._cloud_model,
                 user=user,
                 system=system,
@@ -269,8 +311,9 @@ class AdvisorsAgent(LocalCloudAgent):
                 tools=[build_web_search_tool(ws_max_uses)],
                 max_turns=max_turns,
             )
+            return text, p, c, n_searches, turns, 0.0
         if self._cloud_endpoint == "openai":
-            return self._call_openai_agent(
+            text, p, c, n_searches, turns = self._call_openai_agent(
                 self._cloud_model,
                 user=user,
                 system=system,
@@ -278,8 +321,9 @@ class AdvisorsAgent(LocalCloudAgent):
                 temperature=0.0,
                 max_turns=max_turns,
             )
+            return text, p, c, n_searches, turns, 0.0
         if self._cloud_endpoint == "gemini":
-            return self._call_gemini_agent(
+            text, p, c, n_searches, turns = self._call_gemini_agent(
                 self._cloud_model,
                 user=user,
                 system=system,
@@ -287,6 +331,7 @@ class AdvisorsAgent(LocalCloudAgent):
                 temperature=0.0,
                 max_turns=max_turns,
             )
+            return text, p, c, n_searches, turns, 0.0
         # Genuinely unsupported (openrouter / vllm / unknown). The caller
         # guard should have caught this; raise defensively.
         raise ValueError(

src/openjarvis/agents/hybrid/conductor.py

@@ -46,6 +46,7 @@ from openjarvis.agents.hybrid._base import (
     WEB_SEARCH_COST_PER_CALL,
     LocalCloudAgent,
     build_web_search_tool,
+    tavily_search_context,
     web_search_cfg,
 )
 from openjarvis.agents.hybrid._prices import (
@@ -456,8 +457,14 @@ def _format_worker_pool(workers: List[Dict[str, Any]]) -> str:
     )
 
 
-def _search_capable_indices(workers: List[Dict[str, Any]]) -> List[int]:
+def _search_capable_indices(
+    workers: List[Dict[str, Any]],
+    *,
+    search_backend: str = "provider",
+) -> List[int]:
     """Indices of workers whose endpoint can run server-side web search."""
+    if search_backend == "tavily":
+        return [w["id"] for w in workers]
     return [
         w["id"] for w in workers
         if (w.get("endpoint") or "openai").lower()
@@ -470,6 +477,7 @@ def _build_conductor_prompt(
     workers: List[Dict[str, Any]],
     *,
     web_search_enabled: bool = False,
+    search_backend: str = "provider",
 ) -> str:
     """Build the planner prompt.
 
@@ -485,12 +493,16 @@ def _build_conductor_prompt(
     )
     if not web_search_enabled:
         return base
-    capable = _search_capable_indices(workers)
+    capable = _search_capable_indices(workers, search_backend=search_backend)
     if capable:
         cap_str = ", ".join(str(i) for i in capable)
+        if search_backend == "tavily":
+            capability = "External Tavily search results will be prepended to worker prompts"
+        else:
+            capability = "Only these model indices can perform live web search"
         constraint = (
             "\n\nWEB SEARCH CONSTRAINT:\n"
-            f"Only these model indices can perform live web search: [{cap_str}]. "
+            f"{capability}: [{cap_str}]. "
             "Any step that needs to look up facts, current events, or other "
             "information not reliably known from memory MUST be routed to one "
             "of those indices. Steps routed to any other model can only use "
@@ -550,8 +562,8 @@ def _call_worker(
     *,
     web_search_tool: Optional[Dict[str, Any]] = None,
     web_search_max_uses: int = 8,
-) -> Tuple[str, int, int, bool, int]:
-    """Returns (text, p_tok, c_tok, is_local, n_web_searches).
+) -> Tuple[str, int, int, bool, int, float]:
+    """Returns (text, p_tok, c_tok, is_local, n_web_searches, extra_cost).
 
     ``web_search_tool``: a truthy marker that web_search is enabled for
     this run. When set AND the worker endpoint is search-capable
@@ -565,6 +577,22 @@ def _call_worker(
     max_tok = int(cfg.get("worker_max_tokens", 4096))
     temp = float(cfg.get("worker_temperature", 0.2))
     use_ws = web_search_tool is not None
+    search_backend = str(cfg.get("search_backend", "provider")).lower()
+    extra_cost = 0.0
+    if use_ws and search_backend == "tavily":
+        res = tavily_search_context(
+            prompt,
+            max_results=int(cfg.get("tavily_max_results", 5)),
+        )
+        prompt = (
+            f"Web search results:\n{res['text']}\n\n"
+            f"Using the search results above, answer this request:\n{prompt}"
+        )
+        extra_cost = float(res["cost_usd"])
+        use_ws = False
+        tavily_searches = int(res["n_searches"])
+    else:
+        tavily_searches = 0
 
     if ep == "vllm":
         text, p, c = LocalCloudAgent._call_vllm(
@@ -575,7 +603,7 @@ def _call_worker(
             temperature=temp,
             enable_thinking=False,
         )
-        return text, p, c, True, 0
+        return text, p, c, True, tavily_searches, extra_cost
     if ep == "openai":
         if use_ws:
             text, p, c, n_searches, _ = LocalCloudAgent._call_openai_agent(
@@ -584,14 +612,14 @@ def _call_worker(
                 max_tokens=max_tok,
                 temperature=(1.0 if is_gpt5_family(worker["model"]) else temp),
             )
-            return text, p, c, False, n_searches
+            return text, p, c, False, n_searches, 0.0
         text, p, c = LocalCloudAgent._call_openai(
             worker["model"],
             user=prompt,
             max_tokens=max_tok,
             temperature=(1.0 if is_gpt5_family(worker["model"]) else temp),
         )
-        return text, p, c, False, 0
+        return text, p, c, False, tavily_searches, extra_cost
     if ep == "openrouter":
         # OpenRouter is OpenAI-compatible; the helper handles the
         # base_url + OPENROUTER_API_KEY plumbing. No server-side web
@@ -607,7 +635,7 @@ def _call_worker(
             temperature=temp,
             extra_body=extra_body if isinstance(extra_body, dict) else None,
         )
-        return text, p, c, False, 0
+        return text, p, c, False, tavily_searches, extra_cost
     if ep == "anthropic":
         eff_temp = temp if supports_temperature(worker["model"]) else 0.0
         anthropic_kwargs: Dict[str, Any] = dict(
@@ -620,7 +648,7 @@ def _call_worker(
         text, p, c, n_searches = LocalCloudAgent._call_anthropic(
             worker["model"], **anthropic_kwargs
         )
-        return text, p, c, False, n_searches
+        return text, p, c, False, n_searches or tavily_searches, extra_cost
     if ep == "gemini":
         # Gemini Developer API via google-genai. With web_search on, route
         # through the Google-Search-grounded agent loop; otherwise plain
@@ -632,14 +660,14 @@ def _call_worker(
                 max_tokens=max_tok,
                 temperature=temp,
             )
-            return text, p, c, False, n_searches
+            return text, p, c, False, n_searches, 0.0
         text, p, c = LocalCloudAgent._call_gemini(
             worker["model"],
             user=prompt,
             max_tokens=max_tok,
             temperature=temp,
         )
-        return text, p, c, False, 0
+        return text, p, c, False, tavily_searches, extra_cost
     raise ValueError(f"unsupported worker endpoint: {ep!r}")
 
 
@@ -672,7 +700,7 @@ def _swe_worker_step(
         # backbones today (the loop's tool-call format is Anthropic- or
         # OpenAI-via-vllm-shaped only). Fall back to one-shot for those —
         # SWE-bench-wise they were already weak; this preserves behavior.
-        text, p, c, is_local, n_searches = _call_worker(worker, prompt, cfg)
+        text, p, c, is_local, n_searches, _extra = _call_worker(worker, prompt, cfg)
         return text, p, c, is_local, n_searches, 0
     out = run_swe_agent_loop(
         task,
@@ -753,13 +781,17 @@ class ConductorAgent(LocalCloudAgent):
             and bool(task_meta_early.get("base_commit"))
         )
         ws_enabled, ws_max_uses = web_search_cfg(cfg)
+        search_backend = str(cfg.get("search_backend", "provider")).lower()
         planner_ws = ws_enabled and not swe_mode_early
 
         # 1. Plan — when web_search is on (GAIA), the prompt names which
         # worker indices can actually search, so the planner routes
         # research steps to a search-capable worker.
         user = _build_conductor_prompt(
-            question, workers, web_search_enabled=planner_ws,
+            question,
+            workers,
+            web_search_enabled=planner_ws,
+            search_backend=search_backend,
         )
         plan_text, p_in, p_out = self._call_cloud(
             user=user,
@@ -833,7 +865,7 @@ class ConductorAgent(LocalCloudAgent):
         # memory. Fail loud instead of degrading silently.
         # ``ws_enabled`` / ``ws_max_uses`` computed up front for the planner
         # constraint — reuse them here.
-        if ws_enabled and not swe_mode:
+        if ws_enabled and search_backend != "tavily" and not swe_mode:
             search_workers = [
                 w for w in workers
                 if (w.get("endpoint") or "openai").lower()
@@ -894,7 +926,7 @@ class ConductorAgent(LocalCloudAgent):
                 # may legitimately not need search; see Task-3 planner
                 # constraint that tries to prevent this upfront).
                 if (
-                    ws_enabled and not swe_mode
+                    ws_enabled and search_backend != "tavily" and not swe_mode
                     and worker_ep not in _SEARCH_CAPABLE_WORKER_ENDPOINTS
                 ):
                     self.record_trace_event({
@@ -911,6 +943,7 @@ class ConductorAgent(LocalCloudAgent):
                         ),
                     })
 
+                extra_cost = 0.0
                 if swe_mode:
                     text, w_in, w_out, is_local, n_searches, bash_turns = (
                         _swe_worker_step(
@@ -919,7 +952,9 @@ class ConductorAgent(LocalCloudAgent):
                     )
                     tool_calls += bash_turns
                 else:
-                    text, w_in, w_out, is_local, n_searches = _call_worker(
+                    (
+                        text, w_in, w_out, is_local, n_searches, extra_cost
+                    ) = _call_worker(
                         worker, prompt, cfg,
                         web_search_tool=ws_tool,
                         web_search_max_uses=ws_max_uses,
@@ -930,7 +965,10 @@ class ConductorAgent(LocalCloudAgent):
                 else:
                     tokens_cloud += w_in + w_out
                     cost += self.cost_usd(worker["model"], w_in, w_out)
-                    cost += n_searches * _worker_search_cost_per_call(worker_ep)
+                    if search_backend != "tavily":
+                        cost += n_searches * _worker_search_cost_per_call(worker_ep)
+                if search_backend == "tavily":
+                    cost += extra_cost
                 n_web_searches_total += n_searches
                 tool_calls += n_searches
                 steps.append({
@@ -981,6 +1019,7 @@ class ConductorAgent(LocalCloudAgent):
                 "plan": plan,
                 "fallback_used": fallback_used,
                 "web_search_enabled": ws_enabled,
+                "search_backend": search_backend,
                 "n_web_searches": n_web_searches_total,
                 "parse_attempts": parse_attempts,
                 "workers": [

src/openjarvis/agents/hybrid/minions.py

@@ -48,12 +48,13 @@ from openjarvis.agents.hybrid._base import (
     WEB_SEARCH_COST_PER_CALL,
     LocalCloudAgent,
     build_web_search_tool,
+    tavily_search_context,
     web_search_cfg,
 )
 from openjarvis.agents.hybrid._openai_retry import (
     patch_openai_globally as _patch_openai_globally,
 )
-from openjarvis.agents.hybrid._prices import NO_TEMP_PREFIXES
+from openjarvis.agents.hybrid._prices import NO_TEMP_PREFIXES, default_max_output_tokens
 from openjarvis.agents.hybrid.mini_swe_agent import run_swe_agent_loop
 from openjarvis.core.registry import AgentRegistry
 
@@ -362,6 +363,8 @@ def _prefetch_context(
     cloud_endpoint: str,
     cloud_model: str,
     max_uses: int = 8,
+    search_backend: str = "provider",
+    tavily_max_results: int = 5,
 ) -> Dict[str, Any]:
     """Use Anthropic web_search to fetch real source material the worker can read.
 
@@ -375,6 +378,22 @@ def _prefetch_context(
     out: Dict[str, Any] = {
         "text": "", "tokens": 0, "cost_usd": 0.0, "n_searches": 0,
     }
+    if search_backend == "tavily":
+        try:
+            res = tavily_search_context(question, max_results=tavily_max_results)
+            out.update(
+                text=res["text"],
+                cost_usd=float(res["cost_usd"]),
+                n_searches=int(res["n_searches"]),
+                tokens=0,
+                engine=res.get("engine"),
+                credits=res.get("credits"),
+            )
+            if res.get("error"):
+                out["error"] = res["error"]
+        except Exception as e:
+            out["error"] = f"{type(e).__name__}: {e}"
+        return out
     if cloud_endpoint != "anthropic" or not (question or "").strip():
         return out
     try:
@@ -498,18 +517,22 @@ class MinionsAgent(LocalCloudAgent):
             max_tokens=cfg.get("worker_max_tokens", 4096),
             local=True,
         )
+        cloud_max_tokens = int(
+            cfg.get("cloud_max_tokens")
+            or default_max_output_tokens(self._cloud_model)
+        )
         if self._cloud_endpoint == "openai":
             cloud_client = OpenAIClient(
                 model_name=self._cloud_model,
                 temperature=0.0,
-                max_tokens=4096,
+                max_tokens=cloud_max_tokens,
             )
         elif self._cloud_endpoint == "anthropic":
             # Temperature stripping is handled by the global patch above for Opus 4.7+.
             cloud_client = AnthropicClient(
                 model_name=self._cloud_model,
                 temperature=0.0,
-                max_tokens=4096,
+                max_tokens=cloud_max_tokens,
             )
         elif self._cloud_endpoint == "gemini":
             # The vendored Minion library already special-cases GeminiClient
@@ -520,7 +543,7 @@ class MinionsAgent(LocalCloudAgent):
             cloud_client = GeminiClient(
                 model_name=self._cloud_model,
                 temperature=0.0,
-                max_tokens=4096,
+                max_tokens=cloud_max_tokens,
             )
         else:
             raise ValueError(f"unsupported cloud endpoint: {self._cloud_endpoint!r}")
@@ -560,6 +583,8 @@ class MinionsAgent(LocalCloudAgent):
                 self._cloud_endpoint,
                 self._cloud_model,
                 max_uses=ws_max_uses,
+                search_backend=str(cfg.get("search_backend", "provider")).lower(),
+                tavily_max_results=int(cfg.get("tavily_max_results", 5)),
             )
 
         if prefetch.get("text"):

src/openjarvis/agents/hybrid/runner.py

@@ -38,13 +38,14 @@ except ModuleNotFoundError:
 from openjarvis.agents._stubs import AgentContext, AgentResult
 from openjarvis.agents.hybrid._energy import EnergyCollector
 from openjarvis.agents.hybrid._prompts import format_prompt as _format_prompt
+from openjarvis.core.paths import get_config_dir
 
 PACKAGE_DIR = Path(__file__).parent
 DEFAULT_REGISTRY_DIR = PACKAGE_DIR / "registry"
 DEFAULT_EXPERIMENTS_DIR = Path(
     os.environ.get(
         "OPENJARVIS_HYBRID_EXPERIMENTS_DIR",
-        Path.home() / ".openjarvis" / "experiments" / "hybrid",
+        get_config_dir() / "experiments" / "hybrid",
     )
 )
 DEFAULT_SUBSETS_DIR = DEFAULT_EXPERIMENTS_DIR / "subsets"
@@ -122,6 +123,7 @@ def load_registry(registry_dir: Optional[Path] = None) -> Dict[str, Dict[str, An
 
 # ---------- Bench dispatch ----------
 
+
 def _load_gaia_tasks(n: Optional[int]) -> List[Dict[str, Any]]:
     """GAIA validation. Each task is a dict with `task_id` + `question`."""
     from openjarvis.evals.datasets.gaia import GAIADataset
@@ -137,12 +139,14 @@ def _load_gaia_tasks(n: Optional[int]) -> List[Dict[str, Any]]:
         # id round-trip.
         md = rec.metadata or {}
         task_id = md.get("task_id") or rec.record_id
-        out.append({
-            "task_id": task_id,
-            "question": md.get("question", rec.problem),
-            "reference": rec.reference,
-            "metadata": dict(md),
-        })
+        out.append(
+            {
+                "task_id": task_id,
+                "question": md.get("question", rec.problem),
+                "reference": rec.reference,
+                "metadata": dict(md),
+            }
+        )
     return out
 
 
@@ -155,19 +159,21 @@ def _load_swebench_tasks(n: Optional[int]) -> List[Dict[str, Any]]:
     out: List[Dict[str, Any]] = []
     for rec in ds.iter_records():
         md = rec.metadata or {}
-        out.append({
-            "task_id": md.get("instance_id", rec.record_id),
-            "repo": md.get("repo", ""),
-            "base_commit": md.get("base_commit", ""),
-            "problem_statement": md.get("problem_statement", rec.problem),
-            "hints_text": md.get("hints_text", ""),
-            "test_patch": md.get("test_patch", ""),
-            "FAIL_TO_PASS": md.get("FAIL_TO_PASS", []),
-            "PASS_TO_PASS": md.get("PASS_TO_PASS", []),
-            "version": md.get("version"),
-            "reference": rec.reference,
-            "metadata": dict(md),
-        })
+        out.append(
+            {
+                "task_id": md.get("instance_id", rec.record_id),
+                "repo": md.get("repo", ""),
+                "base_commit": md.get("base_commit", ""),
+                "problem_statement": md.get("problem_statement", rec.problem),
+                "hints_text": md.get("hints_text", ""),
+                "test_patch": md.get("test_patch", ""),
+                "FAIL_TO_PASS": md.get("FAIL_TO_PASS", []),
+                "PASS_TO_PASS": md.get("PASS_TO_PASS", []),
+                "version": md.get("version"),
+                "reference": rec.reference,
+                "metadata": dict(md),
+            }
+        )
     return out
 
 
@@ -206,7 +212,9 @@ def _load_subset_file(subset_path: str) -> Dict[str, Any]:
                 f"subset {p.name} has no 'task_ids' field; got keys {list(data.keys())}"
             )
         return data
-    raise ValueError(f"subset {p.name} must be a list or dict; got {type(data).__name__}")
+    raise ValueError(
+        f"subset {p.name} must be a list or dict; got {type(data).__name__}"
+    )
 
 
 def _apply_subset(
@@ -332,7 +340,11 @@ def _score_swebench(
 
     patch = extract_patch(answer)
     if patch is None:
-        return {"success": False, "score": 0.0, "details": {"reason": "no_patch_extracted"}}
+        return {
+            "success": False,
+            "score": 0.0,
+            "details": {"reason": "no_patch_extracted"},
+        }
 
     record = EvalRecord(
         record_id=task["task_id"],
@@ -368,6 +380,7 @@ def score(
 
 # ---------- Cell run ----------
 
+
 def _cell_dir(cell_name: str, root: Path) -> Path:
     d = root / cell_name
     d.mkdir(parents=True, exist_ok=True)
@@ -440,8 +453,10 @@ def _error_row(task: Dict[str, Any], t0: float, error: str) -> Dict[str, Any]:
     return {
         "task_id": task["task_id"],
         "answer": "",
-        "tokens_local": 0, "tokens_cloud": 0,
-        "cost_usd": 0.0, "latency_s": time.time() - t0,
+        "tokens_local": 0,
+        "tokens_cloud": 0,
+        "cost_usd": 0.0,
+        "latency_s": time.time() - t0,
         "web_search_uses": 0,
         "tool_calls": 0,
         "n_cloud_calls": 0,
@@ -456,11 +471,13 @@ def _run_one_inner(
 ) -> Dict[str, Any]:
     """Run the agent on one task. Returns a hybrid-shape row."""
     prompt = _format_prompt(task)
-    ctx = AgentContext(metadata={
-        "task": task,
-        "task_id": task["task_id"],
-        "log_dir": log_dir,
-    })
+    ctx = AgentContext(
+        metadata={
+            "task": task,
+            "task_id": task["task_id"],
+            "log_dir": log_dir,
+        }
+    )
     t0 = time.time()
     try:
         result: AgentResult = agent.run(prompt, ctx)
@@ -534,11 +551,12 @@ def _run_one(
     if worker.is_alive():
         print(
             f"[timeout] task={task['task_id']} exceeded "
-            f"{task_timeout_s/60:.1f}m — abandoning worker, recording error row",
+            f"{task_timeout_s / 60:.1f}m — abandoning worker, recording error row",
             flush=True,
         )
         return _error_row(
-            task, t0,
+            task,
+            t0,
             f"TaskTimeout: task exceeded the {task_timeout_s:.0f}s hybrid "
             "per-task wall-clock cap (likely a hung network or Modal-harness "
             "call); worker thread abandoned, task left for resume.",
@@ -558,7 +576,7 @@ def _heartbeat(done: int, total: int, row: Dict[str, Any], t_start: float) -> No
     print(
         f"[{done}/{total}] {ok} task={row['task_id']} score={sc_str} "
         f"local={row['tokens_local']} cloud={row['tokens_cloud']} "
-        f"${row['cost_usd']:.3f} {row['latency_s']:.1f}s eta={eta/60:.1f}m",
+        f"${row['cost_usd']:.3f} {row['latency_s']:.1f}s eta={eta / 60:.1f}m",
         flush=True,
     )
 
@@ -644,9 +662,9 @@ def _write_summary(
     summary_path.write_text(json.dumps(summary, indent=2))
     print(
         f"[summary] {cell_name}: n={n_done}/{cell['n']} err={n_err} "
-        f"acc={acc:.3f} cost=${total_cost:.2f} time={wall/60:.1f}m "
-        f"energy={energy_j/1000:.1f}kJ "
-        f"(session +{elapsed/60:.1f}m +{energy_j_session/1000:.1f}kJ, "
+        f"acc={acc:.3f} cost=${total_cost:.2f} time={wall / 60:.1f}m "
+        f"energy={energy_j / 1000:.1f}kJ "
+        f"(session +{elapsed / 60:.1f}m +{energy_j_session / 1000:.1f}kJ, "
         f"processed={n_processed})",
         flush=True,
     )
@@ -664,8 +682,11 @@ def run_cell(
     out_dir = _cell_dir(cell_name, out_root)
     with _cell_lock(out_dir, cell_name):
         _run_cell_locked(
-            cell_name, cell, out_dir,
-            do_score=do_score, resume=resume,
+            cell_name,
+            cell,
+            out_dir,
+            do_score=do_score,
+            resume=resume,
         )
 
 
@@ -727,7 +748,7 @@ def _run_cell_locked(
     mcfg = cell.get("method_cfg") or {}
     task_timeout_s = float(mcfg.get("task_timeout_s", DEFAULT_TASK_TIMEOUT_S))
     if task_timeout_s > 0:
-        print(f"[task-timeout] {task_timeout_s/60:.1f}m per task", flush=True)
+        print(f"[task-timeout] {task_timeout_s / 60:.1f}m per task", flush=True)
 
     agent = _build_agent(cell)
 
@@ -739,18 +760,25 @@ def _run_cell_locked(
 
     def _process(task: Dict[str, Any]) -> None:
         row = _run_one(
-            agent, cell["bench"], task, log_dir,
+            agent,
+            cell["bench"],
+            task,
+            log_dir,
             task_timeout_s=task_timeout_s,
         )
         scored: Optional[Dict[str, Any]] = None
         if do_score and row.get("error") is None:
             try:
                 scored = score(
-                    cell["bench"], task, row["answer"], cell_name=cell_name,
+                    cell["bench"],
+                    task,
+                    row["answer"],
+                    cell_name=cell_name,
                 )
             except Exception as e:
                 scored = {
-                    "success": False, "score": 0.0,
+                    "success": False,
+                    "score": 0.0,
                     "details": {"score_error": str(e)},
                 }
         full_row = {**row, "score": scored}
@@ -789,9 +817,7 @@ def _run_cell_locked(
         while not watchdog_stop.wait(60.0):
             try:
                 cur = (
-                    results_path.stat().st_mtime
-                    if results_path.exists()
-                    else last_seen
+                    results_path.stat().st_mtime if results_path.exists() else last_seen
                 )
             except Exception:
                 cur = last_seen
@@ -830,7 +856,11 @@ def _run_cell_locked(
     watchdog_stop.set()
 
     _write_summary(
-        out_dir, cell_name, cell, tasks, t_start,
+        out_dir,
+        cell_name,
+        cell,
+        tasks,
+        t_start,
         n_processed=len(pending),
         energy_j_session=energy.energy_j_total,
     )
@@ -838,6 +868,7 @@ def _run_cell_locked(
 
 # ---------- CLI ----------
 
+
 def main(argv: Optional[List[str]] = None) -> int:
     p = argparse.ArgumentParser(
         prog="python -m openjarvis.agents.hybrid.runner",
@@ -855,13 +886,18 @@ def main(argv: Optional[List[str]] = None) -> int:
         help="Override experiments output root.",
     )
     p.add_argument("--no-score", action="store_true", help="Skip scoring.")
-    p.add_argument("--no-resume", action="store_true", help="Don't resume from results.jsonl.")
+    p.add_argument(
+        "--no-resume", action="store_true", help="Don't resume from results.jsonl."
+    )
     args = p.parse_args(argv)
 
     reg_dir = Path(args.registry_dir) if args.registry_dir else None
     cells = load_registry(reg_dir)
     if not cells:
-        print(f"[error] no cells found in {reg_dir or DEFAULT_REGISTRY_DIR}", file=sys.stderr)
+        print(
+            f"[error] no cells found in {reg_dir or DEFAULT_REGISTRY_DIR}",
+            file=sys.stderr,
+        )
         return 2
     if args.cell not in cells:
         print(
@@ -871,7 +907,8 @@ def main(argv: Optional[List[str]] = None) -> int:
         return 2
     root = Path(args.root) if args.root else None
     run_cell(
-        args.cell, cells[args.cell],
+        args.cell,
+        cells[args.cell],
         do_score=not args.no_score,
         resume=not args.no_resume,
         root=root,

src/openjarvis/agents/hybrid/skillorchestra.py

@@ -149,6 +149,17 @@ def _build_router_schema(agent_ids: List[str]) -> Dict[str, Any]:
     }
 
 
+def _openai_response_format(schema: Dict[str, Any]) -> Dict[str, Any]:
+    return {
+        "type": "json_schema",
+        "json_schema": {
+            "name": "skillorchestra_route",
+            "schema": schema["format"]["schema"],
+            "strict": True,
+        },
+    }
+
+
 def _parse_router_json(text: str) -> Dict[str, Any]:
     s = (text or "").strip()
     try:
@@ -196,6 +207,70 @@ class SkillOrchestraAgent(LocalCloudAgent):
 
     agent_id = "skillorchestra"
 
+    def _route_call(
+        self,
+        *,
+        question: str,
+        router_sys: str,
+        router_schema: Dict[str, Any],
+        router_max: int,
+    ) -> Tuple[str, int, int]:
+        user = f"Question:\n{question}"
+        if self._cloud_endpoint == "anthropic":
+            kwargs: Dict[str, Any] = {
+                "user": user,
+                "system": router_sys,
+                "max_tokens": router_max,
+                "output_config": router_schema,
+            }
+            if supports_temperature(self._cloud_model):
+                kwargs["temperature"] = 0.0
+            text, r_in, r_out, _ = self._call_anthropic(
+                self._cloud_model,
+                **kwargs,
+            )
+            return text, r_in, r_out
+        if self._cloud_endpoint == "openai":
+            return self._call_openai(
+                self._cloud_model,
+                user=user,
+                system=router_sys,
+                max_tokens=router_max,
+                temperature=0.0,
+                response_format=_openai_response_format(router_schema),
+            )
+        if self._cloud_endpoint == "gemini":
+            return self._call_gemini(
+                self._cloud_model,
+                user=user,
+                system=router_sys,
+                max_tokens=router_max,
+                temperature=0.0,
+            )
+        raise ValueError(
+            f"SkillOrchestra router unsupported cloud_endpoint={self._cloud_endpoint!r}"
+        )
+
+    def _executor_call(
+        self,
+        *,
+        question: str,
+        max_tokens: int,
+    ) -> Tuple[str, int, int]:
+        if self._cloud_endpoint == "anthropic":
+            text, w_in, w_out, _ = self._call_anthropic(
+                self._cloud_model,
+                user=question,
+                max_tokens=max_tokens,
+                temperature=0.0,
+            )
+            return text, w_in, w_out
+        return self._call_cloud(
+            user=question,
+            max_tokens=max_tokens,
+            temperature=0.0,
+        )
+
     def _is_soft_failure(self, exc: BaseException) -> Optional[str]:
         # Empty/unbalanced router JSON — treat as soft failure to match the
         # hybrid adapter's behavior (matches `err=1` rows in the n=30 cell).
@@ -220,33 +295,13 @@ class SkillOrchestraAgent(LocalCloudAgent):
         router_sys = _build_router_sys(competence, cost)
         router_schema = _build_router_schema(agent_ids)
 
-        # 1. Route — Anthropic only (output_config schema is Anthropic-specific
-        # in the hybrid adapter). If you need OpenAI routing, swap the prompt
-        # to JSON-mode and bypass output_config.
-        if self._cloud_endpoint != "anthropic":
-            raise ValueError(
-                "SkillOrchestra router requires cloud_endpoint='anthropic'; "
-                f"got {self._cloud_endpoint!r}"
-            )
         router_max = int(cfg.get("router_max_tokens", 1024))
-        # Strip temperature for Opus 4.7+; Anthropic's output_config does the schema.
-        if supports_temperature(self._cloud_model):
-            router_text, r_in, r_out, _ = self._call_anthropic(
-                self._cloud_model,
-                user=f"Question:\n{question}",
-                system=router_sys,
-                max_tokens=router_max,
-                temperature=0.0,
-                output_config=router_schema,
-            )
-        else:
-            router_text, r_in, r_out, _ = self._call_anthropic(
-                self._cloud_model,
-                user=f"Question:\n{question}",
-                system=router_sys,
-                max_tokens=router_max,
-                output_config=router_schema,
-            )
+        router_text, r_in, r_out = self._route_call(
+            question=question,
+            router_sys=router_sys,
+            router_schema=router_schema,
+            router_max=router_max,
+        )
 
         decision = _parse_router_json(router_text)
         skill_weights: Dict[str, float] = decision.get("skill_weights") or {}
@@ -329,11 +384,9 @@ class SkillOrchestraAgent(LocalCloudAgent):
                 tokens_cloud += out["tokens_in"] + out["tokens_out"]
                 run_cost += out["cost_usd"]
             else:
-                ans, w_in, w_out, _ = self._call_anthropic(
-                    self._cloud_model,
-                    user=question,
+                ans, w_in, w_out = self._executor_call(
+                    question=question,
                     max_tokens=int(cfg.get("cloud_max_tokens", 4096)),
-                    temperature=0.0,
                 )
                 tokens_cloud += w_in + w_out
                 run_cost += self.cost_usd(self._cloud_model, w_in, w_out)

src/openjarvis/agents/hybrid/skillorchestra/orchestrator.py

@@ -31,7 +31,14 @@ from .stage_router import (
     get_routing_strategy,
     parse_skill_analysis,
 )
-from .tools import anthropic_tools, openai_tools, run_answer, run_code, run_search
+from .tools import (
+    anthropic_tools,
+    gemini_tools,
+    openai_tools,
+    run_answer,
+    run_code,
+    run_search,
+)
 
 # tool name -> routing stage (stage_router uses "reasoning" for code).
 _TOOL_STAGE = {
@@ -115,10 +122,49 @@ def _orchestrate_step(
         u = resp.usage
         p = getattr(u, "prompt_tokens", 0) if u else 0
         c = getattr(u, "completion_tokens", 0) if u else 0
+    elif endpoint == "gemini":
+        from google import genai
+        from google.genai import types
+
+        client = genai.Client(
+            http_options=types.HttpOptions(timeout=600_000)
+        )
+        cfg = types.GenerateContentConfig(
+            temperature=1.0,
+            max_output_tokens=max_tokens,
+            tools=[types.Tool(function_declarations=gemini_tools())],
+        )
+        resp = client.models.generate_content(
+            model=model,
+            contents=user,
+            config=cfg,
+        )
+        text = (resp.text or "") if hasattr(resp, "text") else ""
+        tool_calls = []
+        try:
+            parts = resp.candidates[0].content.parts or []
+        except Exception:  # noqa: BLE001
+            parts = []
+        for part in parts:
+            fc = getattr(part, "function_call", None)
+            if fc is None:
+                continue
+            name = getattr(fc, "name", None)
+            if not isinstance(name, str) or not name:
+                continue
+            args = getattr(fc, "args", None) or {}
+            try:
+                args = dict(args)
+            except Exception:  # noqa: BLE001
+                args = {}
+            tool_calls.append({"name": name, "input": args})
+        um = getattr(resp, "usage_metadata", None)
+        p = int(getattr(um, "prompt_token_count", 0) or 0) if um else 0
+        c = int(getattr(um, "candidates_token_count", 0) or 0) if um else 0
     else:
         raise ValueError(
             f"orchestrator endpoint {endpoint!r} unsupported — route the "
-            "orchestrator through anthropic/openai (set method_cfg."
+            "orchestrator through anthropic/openai/gemini (set method_cfg."
             "orchestrator_endpoint)."
         )
 
@@ -192,6 +238,8 @@ def run_orchestrator(
     code_timeout = int(cfg.get("code_timeout_s", 60))
     answer_max_tokens = int(cfg.get("answer_max_tokens", 40000))
     ws_max_uses = int(cfg.get("web_search_max_uses", 5))
+    search_backend = str(cfg.get("search_backend", "provider")).lower()
+    tavily_max_results = int(cfg.get("tavily_max_results", 5))
 
     # The orchestrator model: a fixed model per run (the original's
     # MODEL_NAME). Defaults to the cell's cloud model when that endpoint
@@ -203,7 +251,7 @@ def run_orchestrator(
     orch_model = (cfg.get("orchestrator_model")
                   or cfg.get("router_model")
                   or agent._cloud_model)
-    if orch_endpoint not in ("anthropic", "openai"):
+    if orch_endpoint not in ("anthropic", "openai", "gemini"):
         orch_endpoint, orch_model = "anthropic", "claude-opus-4-7"
     orch_max_tokens = int(cfg.get("orchestrator_max_tokens", 4096))
 
@@ -306,6 +354,8 @@ def run_orchestrator(
                 res = run_search(
                     agent, spec, context_str=context_str, problem=problem,
                     retriever_url=retriever_url, web_search_max_uses=ws_max_uses,
+                    search_backend=search_backend,
+                    tavily_max_results=tavily_max_results,
                 )
                 docs = res["search_results_data"]
                 joined = "\n---\n".join(d for d in docs if d)[:char_cap]

src/openjarvis/agents/hybrid/skillorchestra/tools.py

@@ -27,6 +27,7 @@ from .._base import (
     OPENAI_WEB_SEARCH_COST_PER_CALL,
     WEB_SEARCH_COST_PER_CALL,
     build_web_search_tool,
+    tavily_search_context,
 )
 from .pool import ModelSpec, call_alias
 
@@ -110,6 +111,26 @@ def openai_tools() -> List[Dict[str, Any]]:
     return out
 
 
+def gemini_tools() -> List[Dict[str, Any]]:
+    """The 3 orchestrator tools in Gemini function-declaration shape."""
+    out = []
+    for name, desc in (
+        ("search", _SEARCH_DESC),
+        ("enhance_reasoning", _CODE_DESC),
+        ("answer", _ANSWER_DESC),
+    ):
+        out.append({
+            "name": name,
+            "description": desc,
+            "parameters": {
+                "type": "object",
+                "properties": {"model": _model_prop(name)},
+                "required": ["model"],
+            },
+        })
+    return out
+
+
 # ---------------------------------------------------------------------------
 # enhance_reasoning / code  —  eval_frames.py:659-812
 # ---------------------------------------------------------------------------
@@ -256,6 +277,8 @@ def run_search(
     retriever_url: Optional[str] = None,
     topk: int = 150,
     web_search_max_uses: int = 5,
+    search_backend: str = "provider",
+    tavily_max_results: int = 5,
 ) -> Dict[str, Any]:
     """Write a search query with ``spec``, then retrieve documents.
 
@@ -283,7 +306,12 @@ def run_search(
     contents: List[str] = []
     search_uses = 0
 
-    if retriever_url:
+    if search_backend == "tavily":
+        res = tavily_search_context(query, max_results=tavily_max_results)
+        contents.append(res["text"])
+        search_uses = int(res["n_searches"])
+        cost += float(res["cost_usd"])
+    elif retriever_url:
         # Faithful path — the original FAISS retriever service.
         import requests
 

src/openjarvis/agents/hybrid/toolorchestra.py

@@ -16,7 +16,8 @@ Two modes, gated by ``method_cfg.orchestrator_mode``:
   (``answer-1``, ``reasoner-2``, ``search-3``, …) is mapped to a real
   backend through ``EXPERT_MODEL_MAPPING`` — by default the frontier
   Anthropic worker for `*-1` slots, gpt-5-mini for `*-2`, local Qwen
-  for `*-3`. Search routes to the Anthropic server-side web_search.
+  for `*-3`. Search routes to the configured provider's server-side
+  web-search helper when available.
 
   We do NOT reproduce the upstream Tavily / FAISS-wiki retriever, the
   code-interpreter sandbox, or the multi-vLLM mix (Llama-3.3-70B,
@@ -43,8 +44,8 @@ Prompted-mode pipeline:
    prompt; fallback to strongest worker on parse failure.
 
 Workers come from ``cfg["workers"]`` or a sensible default pool (local
-Qwen if vLLM up, plus a web-search tool via Anthropic, Opus 4.7,
-gpt-5-mini).
+Qwen if vLLM up, plus provider-native web search, the configured frontier
+cloud model, and gpt-5-mini).
 """
 
 from __future__ import annotations
@@ -59,8 +60,11 @@ from typing import Any, Dict, List, Optional, Tuple
 from openjarvis.agents._stubs import AgentContext
 from openjarvis.agents.hybrid._base import (
     ANTHROPIC_WEB_SEARCH_TOOL,
+    GEMINI_SEARCH_COST_PER_CALL,
+    OPENAI_WEB_SEARCH_COST_PER_CALL,
     WEB_SEARCH_COST_PER_CALL,
     LocalCloudAgent,
+    tavily_search_context,
 )
 from openjarvis.agents.hybrid._prices import (
     PRICES,
@@ -197,10 +201,23 @@ def _expert_for(slot: str, local_model: Optional[str],
                                   cost tier for mid OpenAI calls)
       - `*-3` (local tier)     -> local vLLM (`local_model`)
       - `answer-math-*`        -> same tiers as the numeric suffix
-      - `search-*`             -> always the Anthropic web_search tool (the
-                                  upstream uses Tavily; we have web_search)
+      - `search-*`             -> provider-native web search when the cloud
+                                  endpoint supports it; otherwise Anthropic
     """
     if slot.startswith("search"):
+        ep = (cloud_endpoint or "anthropic").lower()
+        if ep == "openai":
+            return {
+                "name": f"search:{slot}",
+                "type": "openai-web-search",
+                "model": cloud_model,
+            }
+        if ep == "gemini":
+            return {
+                "name": f"search:{slot}",
+                "type": "gemini-web-search",
+                "model": cloud_model,
+            }
         return {
             "name": f"search:{slot}",
             "type": "anthropic-web-search",
@@ -340,21 +357,18 @@ def _paper_expert_for(
 
 # ---- Tavily + Modal helpers -------------------------------------------------
 
-def _call_tavily_search(query: str, max_results: int = 5) -> Tuple[str, int, int]:
-    """One-shot Tavily search. Returns (text, p_tok=0, c_tok=0).
+def _call_tavily_search(
+    query: str,
+    max_results: int = 5,
+) -> Tuple[str, int, int, float, int]:
+    """One-shot Tavily search. Returns (text, p_tok=0, c_tok=0, cost, uses).
 
     Token counts are reported as zero (no LLM was billed); the OpenJarvis
     accounting layer separately tallies tool-call counts. Falls back to
     DuckDuckGo if Tavily is unreachable (see ``WebSearchTool``).
     """
-    from openjarvis.tools.web_search import WebSearchTool
-
-    tool = WebSearchTool(max_results=max_results)
-    res = tool.execute(query=query, max_results=max_results)
-    text = res.content or ""
-    if not res.success and not text:
-        text = "(no results)"
-    return text, 0, 0
+    res = tavily_search_context(query, max_results=max_results)
+    return res["text"], 0, 0, float(res["cost_usd"]), int(res["n_searches"])
 
 
 _MODAL_APP_NAME = "openjarvis-toolorchestra-sandbox"
@@ -674,13 +688,25 @@ def _default_pool(
                 "concise extraction, formatting, arithmetic on given data."
             ),
         })
+    if ep == "openai":
+        search_type = "openai-web-search"
+        search_model = cloud_model
+        search_desc = "OpenAI hosted web search on the configured frontier model."
+    elif ep == "gemini":
+        search_type = "gemini-web-search"
+        search_model = cloud_model
+        search_desc = "Gemini Google Search grounding on the configured frontier model."
+    else:
+        search_type = "anthropic-web-search"
+        search_model = _DEFAULT_WEB_SEARCH_MODEL
+        search_desc = "Anthropic server-side web_search."
     pool.append({
         "id": len(pool),
         "name": "web-search",
-        "type": "anthropic-web-search",
-        "model": "claude-haiku-4-5",
+        "type": search_type,
+        "model": search_model,
         "description": (
-            "Anthropic server-side web_search. Use for facts that need a lookup "
+            f"{search_desc} Use for facts that need a lookup "
             "(recent events, rare names/dates, niche sources). Returns a digest."
         ),
     })
@@ -717,8 +743,13 @@ def _default_pool(
 #   `modal-python`   — One-shot Python exec in a fresh Modal Sandbox (the
 #                      paper's "Python sandbox" inside `enhance_reasoning`).
 _TOOLORCH_VALID_TYPES = (
-    "vllm", "openai", "anthropic", "anthropic-web-search", "gemini",
-    "tavily-search", "openrouter", "modal-python",
+    "vllm", "openai", "anthropic", "anthropic-web-search",
+    "openai-web-search", "gemini", "gemini-web-search", "tavily-search",
+    "openrouter", "modal-python",
+)
+_TOOLORCH_SEARCH_TYPES = (
+    "anthropic-web-search", "openai-web-search", "gemini-web-search",
+    "tavily-search",
 )
 
 # Default model used when an `anthropic-web-search` entry omits `model`.
@@ -739,10 +770,12 @@ def _resolve_worker_pool(
     the override is absent.
 
     Each user-supplied entry must be a dict with keys ``id``, ``name``,
-    ``type``, and (for non-search types) ``model``. ``type`` must be one
-    of ``vllm`` / ``openai`` / ``anthropic`` / ``anthropic-web-search``.
-    ``anthropic-web-search`` entries may omit ``model`` — it defaults to
-    ``claude-haiku-4-5``.
+    ``type``, and (for non-search types) ``model``. Search worker types are
+    ``anthropic-web-search``, ``openai-web-search``, ``gemini-web-search``,
+    and ``tavily-search``. ``anthropic-web-search`` entries may omit
+    ``model`` — it defaults to ``claude-haiku-4-5``. OpenAI and Gemini
+    search workers default to the configured cloud model. Tavily does not
+    require a model.
 
     Substitution: ``model = "$local"`` (or ``"<local>"``) resolves to
     ``local_model``; ``model = "$cloud"`` / ``"<cloud>"`` to ``cloud_model``.
@@ -804,14 +837,24 @@ def _resolve_worker_pool(
         elif isinstance(model, str) and model in ("$cloud", "<cloud>"):
             model = cloud_model
             entry["model"] = model
-        if wtype == "anthropic-web-search":
+        if wtype in _TOOLORCH_SEARCH_TYPES:
             if model in (None, ""):
-                model = _DEFAULT_WEB_SEARCH_MODEL
+                if wtype == "anthropic-web-search":
+                    model = _DEFAULT_WEB_SEARCH_MODEL
+                elif wtype in ("openai-web-search", "gemini-web-search"):
+                    model = cloud_model
+                else:
+                    model = wtype
                 entry["model"] = model
             elif not isinstance(model, str):
                 raise ValueError(
                     f"Invalid worker_pool entry [{wid}]: 'model' must be a string when set"
                 )
+            if wtype in ("openai-web-search", "gemini-web-search") and model not in PRICES:
+                raise ValueError(
+                    f"Invalid worker_pool entry [{wid}]: model {model!r} "
+                    f"is not in PRICES (known: {sorted(PRICES)})"
+                )
             # Search workers don't satisfy the "needs a solver" requirement.
         else:
             if not isinstance(model, str) or not model:
@@ -843,7 +886,7 @@ def _resolve_worker_pool(
     if not has_non_search:
         raise ValueError(
             "Invalid worker_pool entry [-]: worker_pool must contain at least "
-            "one non-search worker (vllm / openai / anthropic)"
+            "one non-search worker (vllm / openai / anthropic / gemini)"
         )
     return resolved
 
@@ -910,13 +953,31 @@ def _call_worker(
         )
         extra = n_searches * WEB_SEARCH_COST_PER_CALL
         return text, p, c, False, extra, n_searches
+    if wtype == "openai-web-search":
+        eff_temp = 1.0 if is_gpt5_family(worker["model"]) else temp
+        text, p, c, n_searches, _ = LocalCloudAgent._call_openai_agent(
+            worker["model"],
+            user=prompt,
+            max_tokens=max(max_tok, 16384) if is_gpt5_family(worker["model"]) else max_tok,
+            temperature=eff_temp,
+        )
+        extra = n_searches * OPENAI_WEB_SEARCH_COST_PER_CALL
+        return text, p, c, False, extra, n_searches
+    if wtype == "gemini-web-search":
+        text, p, c, n_searches, _ = LocalCloudAgent._call_gemini_agent(
+            worker["model"],
+            user=prompt,
+            max_tokens=max_tok,
+            temperature=temp,
+        )
+        extra = n_searches * GEMINI_SEARCH_COST_PER_CALL
+        return text, p, c, False, extra, n_searches
     if wtype == "tavily-search":
-        # Tavily costs are flat per call; charge `WEB_SEARCH_COST_PER_CALL`
-        # for parity with the Anthropic web-search worker. One call = one
-        # "n_search" for accounting.
         max_results = int(cfg.get("tavily_max_results", 5))
-        text, p, c = _call_tavily_search(str(prompt), max_results=max_results)
-        return text, p, c, False, WEB_SEARCH_COST_PER_CALL, 1
+        text, p, c, extra, n_searches = _call_tavily_search(
+            str(prompt), max_results=max_results,
+        )
+        return text, p, c, False, extra, n_searches
     if wtype == "openrouter":
         text, p, c = LocalCloudAgent._call_openrouter(
             worker["model"],
@@ -951,7 +1012,7 @@ def _swe_call_worker(
     caller can surface ``tool_calls`` per row. Fallbacks to one-shot
     workers return 0 bash turns (no agent loop ran)."""
     wtype = worker.get("type", "openai")
-    if wtype == "anthropic-web-search":
+    if wtype in _TOOLORCH_SEARCH_TYPES:
         # Search workers stay one-shot.
         text, p, c, is_local, extra, n_searches = _call_worker(worker, prompt, cfg)
         return text, p, c, is_local, extra, n_searches, 0
@@ -1197,7 +1258,8 @@ class ToolOrchestraAgent(LocalCloudAgent):
                 # Search workers are excluded — they answer fact-lookup
                 # questions, not synthesis.
                 non_search = [
-                    w for w in workers if w.get("type") != "anthropic-web-search"
+                    w for w in workers
+                    if w.get("type") not in _TOOLORCH_SEARCH_TYPES
                 ] or workers
                 worker = max(
                     non_search,

src/openjarvis/agents/manager.py

@@ -11,10 +11,11 @@ import logging
 import sqlite3
 import time
 import uuid
-from pathlib import Path
 from typing import Any, Dict, List, Optional
 from uuid import uuid4
 
+from openjarvis.core.paths import get_config_dir
+
 logger = logging.getLogger(__name__)
 
 _CREATE_AGENTS = """\
@@ -537,7 +538,7 @@ class AgentManager:
             pass
 
         # User templates
-        user_dir = Path("~/.openjarvis/templates").expanduser()
+        user_dir = get_config_dir() / "templates"
         if user_dir.is_dir():
             for f in user_dir.glob("*.toml"):
                 try:

src/openjarvis/agents/morning_digest.py

@@ -13,6 +13,7 @@ from typing import Any, List, Optional
 
 from openjarvis.agents._stubs import AgentContext, AgentResult, ToolUsingAgent
 from openjarvis.agents.digest_store import DigestArtifact, DigestStore
+from openjarvis.core.paths import get_config_dir
 from openjarvis.core.registry import AgentRegistry
 from openjarvis.core.types import Message, Role, ToolCall
 
@@ -21,7 +22,7 @@ def _load_persona(persona_name: str) -> str:
     """Load a persona prompt file by name."""
     search_paths = [
         Path("configs/openjarvis/prompts/personas") / f"{persona_name}.md",
-        Path.home() / ".openjarvis" / "prompts" / "personas" / f"{persona_name}.md",
+        get_config_dir() / "prompts" / "personas" / f"{persona_name}.md",
     ]
     for p in search_paths:
         if p.exists():
@@ -202,7 +203,7 @@ class MorningDigestAgent(ToolUsingAgent):
         tts_text = re.sub(r"\*{1,2}([^*]+)\*{1,2}", r"\1", tts_text)
         tts_text = tts_text.strip()
 
-        output_dir = str(Path.home() / ".openjarvis" / "digests")
+        output_dir = str(get_config_dir() / "digests")
         tts_call = ToolCall(
             id="digest-tts-1",
             name="text_to_speech",

src/openjarvis/agents/proactive_agent.py

@@ -43,6 +43,7 @@ from typing import Any, Dict, List, Optional, Set
 
 from openjarvis.agents._stubs import AgentContext, AgentResult, ToolUsingAgent
 from openjarvis.core.config import load_config
+from openjarvis.core.paths import get_config_dir
 from openjarvis.core.registry import AgentRegistry
 from openjarvis.core.types import Message, Role, ToolCall
 from openjarvis.tools.approval_store import (
@@ -342,8 +343,8 @@ class ProactiveAgent(ToolUsingAgent):
         return self._approval_store
 
     def _build_system_prompt(self) -> str:
-        user_md = _load_md_file(Path.home() / ".openjarvis" / "USER.md")
-        memory_md = _load_md_file(Path.home() / ".openjarvis" / "MEMORY.md")
+        user_md = _load_md_file(get_config_dir() / "USER.md")
+        memory_md = _load_md_file(get_config_dir() / "MEMORY.md")
         now = datetime.now()
         context_block = ""
         if user_md or memory_md:

src/openjarvis/agents/prompt_loader.py

@@ -15,16 +15,17 @@ from __future__ import annotations
 
 import json
 import logging
-import os
 from pathlib import Path
 from typing import Any
 
+from openjarvis.core.paths import get_config_dir
+
 logger = logging.getLogger(__name__)
 
 
 def _openjarvis_home() -> Path:
-    """Resolve $OPENJARVIS_HOME, defaulting to ~/.openjarvis."""
-    return Path(os.environ.get("OPENJARVIS_HOME", "~/.openjarvis")).expanduser()
+    """Resolve the OpenJarvis root, honoring OPENJARVIS_HOME / XDG_DATA_HOME."""
+    return get_config_dir()
 
 
 def load_system_prompt_override(agent_name: str) -> str | None:

src/openjarvis/channels/imessage_daemon.py

@@ -18,11 +18,13 @@ import time
 from pathlib import Path
 from typing import Any, Dict, List
 
+from openjarvis.core.paths import get_config_dir
+
 logger = logging.getLogger(__name__)
 
 _DEFAULT_DB_PATH = str(Path.home() / "Library" / "Messages" / "chat.db")
 _POLL_INTERVAL = 5
-_PID_FILE = str(Path.home() / ".openjarvis" / "imessage-agent.pid")
+_PID_FILE = str(get_config_dir() / "imessage-agent.pid")
 
 
 def poll_new_messages(

src/openjarvis/channels/slack_daemon.py

@@ -14,9 +14,11 @@ import sys
 from pathlib import Path
 from typing import Any
 
+from openjarvis.core.paths import get_config_dir
+
 logger = logging.getLogger(__name__)
 
-_PID_FILE = str(Path.home() / ".openjarvis" / "slack-daemon.pid")
+_PID_FILE = str(get_config_dir() / "slack-daemon.pid")
 
 
 def _to_slack_fmt(text: str) -> str:

src/openjarvis/channels/whatsapp_baileys.py

@@ -22,6 +22,7 @@ from openjarvis.channels._stubs import (
     ChannelStatus,
 )
 from openjarvis.core.events import EventBus, EventType
+from openjarvis.core.paths import get_config_dir
 from openjarvis.core.registry import ChannelRegistry
 
 logger = logging.getLogger(__name__)
@@ -38,7 +39,7 @@ if not _BRIDGE_SRC.exists():
     )
 
 # Default runtime directory (npm install + auth state).
-_DEFAULT_RUNTIME_DIR = Path.home() / ".openjarvis" / "whatsapp_baileys_bridge"
+_DEFAULT_RUNTIME_DIR = get_config_dir() / "whatsapp_baileys_bridge"
 
 
 @ChannelRegistry.register("whatsapp_baileys")

src/openjarvis/cli/_version_check.py

@@ -9,9 +9,11 @@ import sys
 import time
 from pathlib import Path
 
+from openjarvis.core.paths import get_config_dir
+
 logger = logging.getLogger(__name__)
 
-_CACHE_PATH = Path("~/.openjarvis/version-check.json").expanduser()
+_CACHE_PATH = get_config_dir() / "version-check.json"
 _CACHE_TTL = 86400  # 24 hours
 _PYPI_API = "https://pypi.org/pypi/openjarvis/json"
 
@@ -21,7 +23,7 @@ def _config_path() -> Path:
     override = os.environ.get("OPENJARVIS_CONFIG")
     if override:
         return Path(override).expanduser()
-    return Path("~/.openjarvis/config.toml").expanduser()
+    return get_config_dir() / "config.toml"
 
 
 # Commands that surface the "new version available" nudge. We deliberately

src/openjarvis/cli/agent_cmd.py

@@ -12,15 +12,12 @@ from rich.table import Table
 
 def _get_manager():
     """Get or create the AgentManager singleton."""
-    from pathlib import Path
-
     from openjarvis.agents.manager import AgentManager
     from openjarvis.core.config import load_config
+    from openjarvis.core.paths import get_config_dir
 
     config = load_config()
-    db_path = config.agent_manager.db_path or str(
-        Path("~/.openjarvis/agents.db").expanduser()
-    )
+    db_path = config.agent_manager.db_path or str(get_config_dir() / "agents.db")
     return AgentManager(db_path=db_path)
 
 
@@ -273,6 +270,7 @@ def search(agent_id: str, query: str, limit: int) -> None:
     console = Console(stderr=True)
     try:
         from openjarvis.core.config import load_config
+        from openjarvis.core.paths import get_config_dir
         from openjarvis.traces.store import TraceStore
 
         config = load_config()
@@ -281,7 +279,7 @@ def search(agent_id: str, query: str, limit: int) -> None:
         if not agent:
             console.print(f"[red]Agent not found: {agent_id}[/red]")
             return
-        store = TraceStore(config.traces.db_path or "~/.openjarvis/traces.db")
+        store = TraceStore(config.traces.db_path or str(get_config_dir() / "traces.db"))
         results = store.search(query, agent=agent["name"], limit=limit)
         if not results:
             console.print("[dim]No results.[/dim]")
@@ -545,8 +543,7 @@ def run_agent(agent_id):
     updated = manager.get_agent(agent_id)
     runs = updated.get("total_runs", 0)
     console.print(
-        f"[green]✓[/green] Tick complete. "
-        f"Status: {updated['status']}, runs: {runs}"
+        f"[green]✓[/green] Tick complete. Status: {updated['status']}, runs: {runs}"
     )
 
     # Print the agent's actual output. summary_memory holds the latest tick's
@@ -662,6 +659,7 @@ def trace(agent_id, run_number, limit):
     import datetime
 
     from openjarvis.core.config import load_config
+    from openjarvis.core.paths import get_config_dir
     from openjarvis.traces.store import TraceStore
 
     manager = _get_manager()
@@ -671,7 +669,7 @@ def trace(agent_id, run_number, limit):
         raise SystemExit(1)
 
     config = load_config()
-    store = TraceStore(config.traces.db_path or "~/.openjarvis/traces.db")
+    store = TraceStore(config.traces.db_path or str(get_config_dir() / "traces.db"))
     traces = store.list_traces(agent=agent_id, limit=limit)
 
     if not traces:
@@ -841,8 +839,8 @@ def ask(agent_id, message, auto_approve):
     if auto_approve:
         executor._confirm_callback = lambda _prompt: True
     else:
-        executor._confirm_callback = (
-            lambda prompt: click.confirm(f"\n{prompt}", default=False)
+        executor._confirm_callback = lambda prompt: click.confirm(
+            f"\n{prompt}", default=False
         )
     # Run the tick with a live trace rather than blocking in silence — the
     # message we just queued is consumed as this tick's input, so the user

src/openjarvis/cli/config_cmd.py

@@ -276,6 +276,36 @@ def hardware() -> None:
 config.add_command(show_group, "show")
 
 
+@config.command("path")
+def show_path() -> None:
+    """Print the resolved OpenJarvis directories (home, config, cache).
+
+    All OpenJarvis state lives under a single root, resolved in priority
+    order: ``$OPENJARVIS_HOME`` > ``$XDG_DATA_HOME/openjarvis`` >
+    ``~/.openjarvis``. Use this to confirm where your data is stored after
+    setting an override.
+    """
+    from openjarvis.core.paths import get_cache_dir, get_config_dir, get_config_path
+
+    console = Console(stderr=True)
+    home = get_config_dir()
+    override = (
+        "OPENJARVIS_HOME"
+        if os.environ.get("OPENJARVIS_HOME")
+        else "XDG_DATA_HOME"
+        if os.environ.get("XDG_DATA_HOME")
+        else "default (~/.openjarvis)"
+    )
+    table = Table(show_header=True, header_style="bold")
+    table.add_column("Directory")
+    table.add_column("Path", style="cyan")
+    table.add_row("Home (root)", str(home))
+    table.add_row("Config file", str(get_config_path()))
+    table.add_row("Cache", str(get_cache_dir()))
+    console.print(table)
+    console.print(f"[dim]Resolved via: {override}[/dim]")
+
+
 def _probe_engine_host(url: str, console: Console) -> None:
     """Probe an engine host URL and print reachability status."""
     try:

src/openjarvis/cli/doctor_cmd.py

@@ -354,7 +354,9 @@ def doctor(as_json: bool) -> None:
 
     # Background tasks section
     from openjarvis.cli._bg_state import get_status
+    from openjarvis.core.paths import get_config_dir
 
+    scripts_dir = get_config_dir() / ".scripts"
     console.print("[bold]Background tasks[/bold]")
     bg = get_status()
     bg_failed = False
@@ -364,8 +366,8 @@ def doctor(as_json: bool) -> None:
     elif bg.rust_extension == "failed":
         console.print(f"  [red]✗[/red] Rust extension: failed — {bg.rust_error[:80]}")
         console.print(
-            "    retry: ~/.openjarvis/.scripts/install-rust.sh && "
-            "~/.openjarvis/.scripts/build-extension.sh"
+            f"    retry: {scripts_dir}/install-rust.sh && "
+            f"{scripts_dir}/build-extension.sh"
         )
         bg_failed = True
     else:
@@ -380,7 +382,7 @@ def doctor(as_json: bool) -> None:
             console.print(f"  [green]✓[/green] {model_id}: ready")
         elif state == "failed":
             console.print(f"  [red]✗[/red] {model_id}: failed")
-            console.print(f"    retry: ~/.openjarvis/.scripts/pull-model.sh {model_id}")
+            console.print(f"    retry: {scripts_dir}/pull-model.sh {model_id}")
             bg_failed = True
         else:
             console.print(f"  [yellow]…[/yellow] {model_id}: downloading")

src/openjarvis/cli/log_config.py

@@ -7,6 +7,7 @@ from logging.handlers import RotatingFileHandler
 from pathlib import Path
 from typing import Optional, Union
 
+from openjarvis.core.paths import get_config_dir
 from openjarvis.security.credential_stripper import CredentialStripper
 
 _stripper = CredentialStripper()
@@ -68,7 +69,7 @@ def setup_logging(
         if log_file is None:
             from openjarvis.security.file_utils import secure_mkdir
 
-            log_dir = Path.home() / ".openjarvis"
+            log_dir = get_config_dir()
             secure_mkdir(log_dir)
             log_file = log_dir / "cli.log"
         file_handler = RotatingFileHandler(

src/openjarvis/cli/scan_cmd.py

@@ -11,6 +11,8 @@ from typing import Callable, List
 
 import click
 
+from openjarvis.core.paths import get_config_dir
+
 # Engine ports that should only be listening on localhost.
 _ENGINE_PORTS = {11434, 8080, 8000, 30000, 1234, 52415, 18181}
 
@@ -132,7 +134,7 @@ class PrivacyScanner:
     def check_icloud_sync(self) -> ScanResult:
         """Check whether ~/.openjarvis is inside iCloud Drive sync scope."""
         try:
-            config_path = Path("~/.openjarvis").expanduser().resolve()
+            config_path = get_config_dir().resolve()
             icloud_path = Path("~/Library/Mobile Documents/").expanduser().resolve()
             if str(config_path).startswith(str(icloud_path)):
                 return ScanResult(

src/openjarvis/cli/serve.py

@@ -11,6 +11,7 @@ from rich.console import Console
 from openjarvis.cli._banner import print_banner
 from openjarvis.core.config import load_config
 from openjarvis.core.events import EventBus
+from openjarvis.core.paths import get_config_dir
 from openjarvis.engine import (
     discover_engines,
     discover_models,
@@ -496,13 +497,9 @@ def serve(
     agent_manager = None
     if config.agent_manager.enabled:
         try:
-            from pathlib import Path
-
             from openjarvis.agents.manager import AgentManager
 
-            am_db = config.agent_manager.db_path or str(
-                Path("~/.openjarvis/agents.db").expanduser()
-            )
+            am_db = config.agent_manager.db_path or str(get_config_dir() / "agents.db")
             # The server owns the scheduler and is the authoritative tick
             # runner — on boot it holds no locks, so it (and only it) sweeps
             # any zombie running→idle left by a previous crash.
@@ -607,9 +604,7 @@ def serve(
         try:
             import tomllib
 
-            _cfg_path = str(
-                __import__("pathlib").Path.home() / ".openjarvis" / "config.toml"
-            )
+            _cfg_path = str(get_config_dir() / "config.toml")
             with open(_cfg_path, "rb") as _f:
                 _raw = tomllib.load(_f)
             api_key = _raw.get("server", {}).get("auth", {}).get("api_key", "")

src/openjarvis/cli/skill_cmd.py

@@ -11,6 +11,7 @@ from rich.table import Table
 
 from openjarvis.core.config import load_config
 from openjarvis.core.events import EventBus
+from openjarvis.core.paths import get_config_dir
 from openjarvis.skills.manager import SkillManager
 
 
@@ -28,12 +29,12 @@ def _get_trace_store():
 
 def _get_discovered_dir() -> Path:
     """Return the directory where discovered skill manifests are written."""
-    return Path("~/.openjarvis/skills/discovered/").expanduser()
+    return get_config_dir() / "skills" / "discovered"
 
 
 def _get_overlay_dir() -> Path:
     """Return the directory where optimization overlays are stored."""
-    return Path("~/.openjarvis/learning/skills/").expanduser()
+    return get_config_dir() / "learning" / "skills"
 
 
 def _get_skill_paths() -> List[Path]:
@@ -41,7 +42,7 @@ def _get_skill_paths() -> List[Path]:
     workspace = Path("./skills")
     if workspace.exists():
         paths.append(workspace)
-    user_dir = Path("~/.openjarvis/skills/").expanduser()
+    user_dir = get_config_dir() / "skills"
     paths.append(user_dir)
     return paths
 
@@ -174,8 +175,10 @@ def _get_resolver(source: str, url: str = ""):
         from openjarvis.skills.sources.github import GitHubResolver
 
         cache = _Path(
-            "~/.openjarvis/skill-cache/github/" + url.rstrip("/").rsplit("/", 1)[-1]
-        ).expanduser()
+            str(get_config_dir() / "skill-cache" / "github")
+            + "/"
+            + url.rstrip("/").rsplit("/", 1)[-1]
+        )
         return GitHubResolver(cache_root=cache, repo_url=url)
     raise click.BadParameter(f"Unknown source: {source!r}")
 

src/openjarvis/core/config.py

@@ -16,6 +16,14 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
+from openjarvis.core.paths import (
+    ConfigurationError,
+    get_cache_dir,
+    get_config_dir,
+    get_config_path,
+    get_data_dir,
+)
+
 if TYPE_CHECKING:
     # Only used by type-checkers (mypy/pyright) for the ``JarvisConfig.mining``
     # field annotation. The runtime import is deferred inside
@@ -33,15 +41,24 @@ except ModuleNotFoundError:
 # Hardware dataclasses
 # ---------------------------------------------------------------------------
 
-DEFAULT_CONFIG_DIR = Path.home() / ".openjarvis"
-DEFAULT_CONFIG_PATH = DEFAULT_CONFIG_DIR / "config.toml"
+# Legacy names, kept for the ~45 modules that import them. They are resolved
+# once at import via the env-aware resolver in ``openjarvis.core.paths`` (the
+# install-script model: ``OPENJARVIS_HOME`` / ``XDG_DATA_HOME`` are set before
+# the process starts). They are real module attributes — not computed lazily —
+# so existing tests can ``monkeypatch.setattr`` them and so dataclass-instance
+# defaults stay consistent. Code that must react to a mid-process env change
+# (or wants the override regardless of import order) should call
+# ``get_config_dir()`` / ``get_config_path()`` directly; the dataclass field
+# defaults below already do this via ``default_factory``.
+DEFAULT_CONFIG_DIR = get_config_dir()
+DEFAULT_CONFIG_PATH = get_config_path()
 
 
 def _ensure_config_dir() -> Path:
     """Ensure the config directory exists with restrictive permissions."""
     from openjarvis.security.file_utils import secure_mkdir
 
-    return secure_mkdir(DEFAULT_CONFIG_DIR)
+    return secure_mkdir(get_config_dir())
 
 
 @dataclass(slots=True)
@@ -742,7 +759,9 @@ class SkillsLearningConfig:
     optimizer: str = "dspy"  # "dspy" or "gepa"
     min_traces_per_skill: int = 20
     optimization_interval_seconds: int = 86400
-    overlay_dir: str = "~/.openjarvis/learning/skills/"
+    overlay_dir: str = field(
+        default_factory=lambda: str(get_config_dir() / "learning" / "skills")
+    )
 
 
 @dataclass(slots=True)
@@ -894,7 +913,7 @@ class StorageConfig:
     """Storage (memory) backend settings."""
 
     default_backend: str = "sqlite"
-    db_path: str = str(DEFAULT_CONFIG_DIR / "memory.db")
+    db_path: str = field(default_factory=lambda: str(get_config_dir() / "memory.db"))
     context_top_k: int = 5
     context_min_score: float = 0.0
     context_max_tokens: int = 2048
@@ -998,7 +1017,7 @@ class TelemetryConfig:
     """Telemetry persistence settings."""
 
     enabled: bool = True
-    db_path: str = str(DEFAULT_CONFIG_DIR / "telemetry.db")
+    db_path: str = field(default_factory=lambda: str(get_config_dir() / "telemetry.db"))
     gpu_metrics: bool = False
     gpu_poll_interval_ms: int = 50
     energy_vendor: str = ""  # auto-detect or force "nvidia"/"amd"/"apple"/"cpu_rapl"
@@ -1023,7 +1042,7 @@ class AnalyticsConfig:
     enabled: bool = True
     host: str = "https://34.231.106.201.sslip.io"
     key: str = "phc_ysKu72QaxzYNmDpHFcesD2ZZAe68zkdWJEKoYYkc5e3n"
-    anon_id_path: str = str(DEFAULT_CONFIG_DIR / "anon_id")
+    anon_id_path: str = field(default_factory=lambda: str(get_config_dir() / "anon_id"))
     flush_interval_seconds: int = 30
     flush_at_size: int = 100
 
@@ -1033,7 +1052,7 @@ class TracesConfig:
     """Trace system settings."""
 
     enabled: bool = True
-    db_path: str = str(DEFAULT_CONFIG_DIR / "traces.db")
+    db_path: str = field(default_factory=lambda: str(get_config_dir() / "traces.db"))
 
 
 @dataclass(slots=True)
@@ -1235,7 +1254,9 @@ class SecurityConfig:
     mode: str = "redact"  # "redact" | "warn" | "block"
     secret_scanner: bool = True
     pii_scanner: bool = True
-    audit_log_path: str = str(DEFAULT_CONFIG_DIR / "audit.db")
+    audit_log_path: str = field(
+        default_factory=lambda: str(get_config_dir() / "audit.db")
+    )
     enforce_tool_confirmation: bool = True
     merkle_audit: bool = True
     signing_key_path: str = ""
@@ -1246,7 +1267,9 @@ class SecurityConfig:
     local_engine_bypass: bool = False
     local_tool_bypass: bool = False
     profile: str = ""
-    vault_key_path: str = str(DEFAULT_CONFIG_DIR / ".vault_key")
+    vault_key_path: str = field(
+        default_factory=lambda: str(get_config_dir() / ".vault_key")
+    )
     capabilities: CapabilitiesConfig = field(default_factory=CapabilitiesConfig)
 
 
@@ -1367,7 +1390,7 @@ class SessionConfig:
     enabled: bool = False
     max_age_hours: float = 24.0
     consolidation_threshold: int = 100
-    db_path: str = str(DEFAULT_CONFIG_DIR / "sessions.db")
+    db_path: str = field(default_factory=lambda: str(get_config_dir() / "sessions.db"))
 
 
 @dataclass(slots=True)
@@ -1385,7 +1408,9 @@ class OperatorsConfig:
     """Operator lifecycle settings."""
 
     enabled: bool = False
-    manifests_dir: str = "~/.openjarvis/operators"
+    manifests_dir: str = field(
+        default_factory=lambda: str(get_config_dir() / "operators")
+    )
     auto_activate: str = ""  # Comma-separated operator IDs
 
 
@@ -1411,7 +1436,7 @@ class OptimizeConfig:
     benchmark: str = ""
     max_samples: int = 50
     judge_model: str = "gpt-5-mini-2025-08-07"
-    db_path: str = str(DEFAULT_CONFIG_DIR / "optimize.db")
+    db_path: str = field(default_factory=lambda: str(get_config_dir() / "optimize.db"))
 
 
 @dataclass(slots=True)
@@ -1419,18 +1444,20 @@ class AgentManagerConfig:
     """Persistent agent manager settings."""
 
     enabled: bool = True
-    db_path: str = str(DEFAULT_CONFIG_DIR / "agents.db")
+    db_path: str = field(default_factory=lambda: str(get_config_dir() / "agents.db"))
 
 
 @dataclass(slots=True)
 class MemoryFilesConfig:
     """Persistent memory-file paths and nudge settings."""
 
-    soul_path: str = "~/.openjarvis/SOUL.md"
-    memory_path: str = "~/.openjarvis/MEMORY.md"
-    user_path: str = "~/.openjarvis/USER.md"
+    soul_path: str = field(default_factory=lambda: str(get_config_dir() / "SOUL.md"))
+    memory_path: str = field(
+        default_factory=lambda: str(get_config_dir() / "MEMORY.md")
+    )
+    user_path: str = field(default_factory=lambda: str(get_config_dir() / "USER.md"))
     nudge_interval: int = 10
-    persona_name: str = ""  # named persona dir under ~/.openjarvis/personas/<name>/
+    persona_name: str = ""  # named persona dir under <config-dir>/personas/<name>/
 
 
 @dataclass(slots=True)
@@ -1469,13 +1496,15 @@ class SkillsConfig:
     """Configuration for agent-authored procedural skills."""
 
     enabled: bool = True
-    skills_dir: str = "~/.openjarvis/skills/"
+    skills_dir: str = field(default_factory=lambda: str(get_config_dir() / "skills"))
     active: str = "*"
     auto_discover: bool = True
     auto_sync: bool = False
     nudge_interval: int = 15
     index_repo: str = "https://github.com/openjarvis/skill-index.git"
-    index_dir: str = "~/.openjarvis/skill-index/"
+    index_dir: str = field(
+        default_factory=lambda: str(get_config_dir() / "skill-index")
+    )
     max_depth: int = 5
     sandbox_dangerous: bool = True
     sources: List[SkillSourceConfig] = field(default_factory=list)
@@ -1781,7 +1810,7 @@ def load_config(path: Optional[Path] = None) -> JarvisConfig:
     elif os.environ.get("OPENJARVIS_CONFIG"):
         config_path = Path(os.environ["OPENJARVIS_CONFIG"]).expanduser().resolve()
     else:
-        config_path = DEFAULT_CONFIG_PATH
+        config_path = get_config_path()
     if config_path.exists():
         with open(config_path, "rb") as fh:
             data = tomllib.load(fh)
@@ -2119,9 +2148,14 @@ __all__ = [
     "BrowserConfig",
     "CapabilitiesConfig",
     "ChannelConfig",
+    "ConfigurationError",
     "DEFAULT_CONFIG_DIR",
     "DEFAULT_CONFIG_PATH",
     "DiscordChannelConfig",
+    "get_cache_dir",
+    "get_config_dir",
+    "get_config_path",
+    "get_data_dir",
     "EmailChannelConfig",
     "EngineConfig",
     "FeishuChannelConfig",

src/openjarvis/core/credentials.py

@@ -10,13 +10,20 @@ import os
 import threading
 from pathlib import Path
 
+from openjarvis.core.paths import get_config_dir
+
 try:
     import tomllib
 except ModuleNotFoundError:
     import tomli as tomllib  # type: ignore[no-redef]
 
 _LOCK = threading.Lock()
-_DEFAULT_PATH = Path.home() / ".openjarvis" / "credentials.toml"
+
+
+def _default_path() -> Path:
+    """Resolve the credentials file under the OpenJarvis root (env-aware)."""
+    return get_config_dir() / "credentials.toml"
+
 
 TOOL_CREDENTIALS: dict[str, list[str]] = {
     "web_search": ["TAVILY_API_KEY"],
@@ -53,7 +60,7 @@ TOOL_CREDENTIALS: dict[str, list[str]] = {
 
 def load_credentials(path: Path | None = None) -> dict[str, dict[str, str]]:
     """Load credentials from TOML file."""
-    p = Path(path) if path else _DEFAULT_PATH
+    p = Path(path) if path else _default_path()
     if not p.exists():
         return {}
     with open(p, "rb") as f:
@@ -75,7 +82,7 @@ def save_credential(
     if not stripped:
         raise ValueError("Credential value must not be empty")
 
-    p = Path(path) if path else _DEFAULT_PATH
+    p = Path(path) if path else _default_path()
     with _LOCK:
         creds = load_credentials(path=p)
         if tool_name not in creds:

src/openjarvis/core/paths.py

@@ -0,0 +1,122 @@
+"""Central, env-aware resolution of OpenJarvis' home directory.
+
+OpenJarvis keeps all of its runtime state (config, databases, caches, logs,
+credentials, skills, recipes, …) under a single root so it never clutters the
+user's home directory beyond one directory. That root is resolved here, with
+the following precedence (highest first):
+
+1. ``$OPENJARVIS_HOME`` — explicit override (also honored by the shell
+   installer, see ``scripts/install/install.sh``).
+2. ``$XDG_DATA_HOME/openjarvis`` — when ``$XDG_DATA_HOME`` is set, follow the
+   XDG Base Directory spec by nesting a single ``openjarvis`` directory under
+   it. We deliberately use ONE directory rather than splitting across XDG
+   config/data/cache so the install tree stays self-contained and relocatable.
+3. ``~/.openjarvis`` — the historical default. With no env vars set, the
+   resolved path is exactly this, so existing installs are untouched.
+
+``config.py`` re-exports :func:`get_config_dir` results through the legacy
+``DEFAULT_CONFIG_DIR``/``DEFAULT_CONFIG_PATH`` names (computed dynamically) so
+the ~45 modules that import those names keep working while honoring the
+override. Modules that previously hardcoded ``Path.home() / ".openjarvis"``
+should call :func:`get_config_dir` (or :func:`get_data_dir` /
+:func:`get_cache_dir`) instead.
+
+Defense in depth: the resolved root must never live inside the OpenJarvis
+source tree (a misconfigured ``$OPENJARVIS_HOME`` pointing at the repo would
+otherwise scatter runtime artifacts into the working tree). This mirrors the
+guard in ``learning/spec_search/storage/paths.py`` and fails loudly per
+REVIEW.md's no-silent-failure discipline.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+_DEFAULT_DIR_NAME = ".openjarvis"
+_XDG_SUBDIR_NAME = "openjarvis"
+
+
+class ConfigurationError(RuntimeError):
+    """Raised when the resolved home directory would violate isolation guarantees."""
+
+
+def _find_source_root() -> Path | None:
+    """Walk upward from this module to find the OpenJarvis source root.
+
+    Returns the directory containing the OpenJarvis ``pyproject.toml`` (the one
+    whose ``name = "openjarvis"``), or ``None`` when running from an installed
+    wheel rather than a source checkout.
+    """
+    here = Path(__file__).resolve()
+    for candidate in (here, *here.parents):
+        py = candidate / "pyproject.toml"
+        if py.exists():
+            try:
+                content = py.read_text(encoding="utf-8")
+            except OSError:
+                continue
+            if 'name = "openjarvis"' in content.lower():
+                return candidate
+    return None
+
+
+def _reject_source_tree(path: Path) -> Path:
+    """Raise if ``path`` resolves inside the OpenJarvis source tree."""
+    source_root = _find_source_root()
+    if source_root is not None:
+        try:
+            path.relative_to(source_root)
+        except ValueError:
+            pass  # Good — not inside the source tree.
+        else:
+            raise ConfigurationError(
+                f"OpenJarvis home ({path}) is inside the source tree "
+                f"({source_root}). OpenJarvis refuses to write runtime state "
+                "inside its own repo. Set OPENJARVIS_HOME (or XDG_DATA_HOME) "
+                "to a directory outside the repo (default: ~/.openjarvis)."
+            )
+    return path
+
+
+def get_config_dir() -> Path:
+    """Resolve OpenJarvis' single root directory, honoring env overrides.
+
+    Precedence: ``$OPENJARVIS_HOME`` > ``$XDG_DATA_HOME/openjarvis`` >
+    ``~/.openjarvis``. The result is always absolute and is rejected if it
+    falls inside the OpenJarvis source tree.
+    """
+    env_home = os.environ.get("OPENJARVIS_HOME")
+    if env_home:
+        resolved = Path(env_home).expanduser().resolve()
+        return _reject_source_tree(resolved)
+
+    xdg_data = os.environ.get("XDG_DATA_HOME")
+    if xdg_data:
+        resolved = (Path(xdg_data).expanduser() / _XDG_SUBDIR_NAME).resolve()
+        return _reject_source_tree(resolved)
+
+    return (Path.home() / _DEFAULT_DIR_NAME).resolve()
+
+
+def get_config_path() -> Path:
+    """Resolve the path to ``config.toml`` under the OpenJarvis root."""
+    return get_config_dir() / "config.toml"
+
+
+def get_data_dir() -> Path:
+    """Resolve the directory for persistent data (databases, blobs, …).
+
+    Consolidated under the single root; identical to :func:`get_config_dir`.
+    Provided as a distinct name so call sites read intentionally.
+    """
+    return get_config_dir()
+
+
+def get_cache_dir() -> Path:
+    """Resolve the directory for regenerable caches (eval datasets, etc.).
+
+    Lives at ``<root>/cache`` so caches stay inside the single OpenJarvis
+    directory instead of scattering across ``~/.cache``.
+    """
+    return get_config_dir() / "cache"

src/openjarvis/evals/datasets/gaia.py

@@ -11,11 +11,12 @@ import shutil
 from pathlib import Path
 from typing import Iterable, List, MutableMapping, Optional, Sequence
 
+from openjarvis.core.paths import get_cache_dir
 from openjarvis.evals.core.dataset import DatasetProvider
 from openjarvis.evals.core.splits import apply_split
 from openjarvis.evals.core.types import EvalRecord
 
-_DEFAULT_CACHE_DIR = Path.home() / ".cache" / "gaia_benchmark"
+_DEFAULT_CACHE_DIR = get_cache_dir() / "gaia_benchmark"
 
 _DEFAULT_INPUT_PROMPT = """Please answer the question below. You should:
 

src/openjarvis/evals/datasets/liveresearch.py

@@ -17,6 +17,7 @@ import subprocess
 from pathlib import Path
 from typing import Any, Dict, Iterable, List, Optional
 
+from openjarvis.core.paths import get_cache_dir
 from openjarvis.evals.core.dataset import DatasetProvider
 from openjarvis.evals.core.splits import apply_split
 from openjarvis.evals.core.types import EvalRecord
@@ -24,7 +25,7 @@ from openjarvis.evals.core.types import EvalRecord
 LOGGER = logging.getLogger(__name__)
 
 LIVERESEARCH_REPO = "https://github.com/Ayanami0730/deep_research_bench.git"
-CACHE_DIR = Path.home() / ".cache" / "liveresearch_bench"
+CACHE_DIR = get_cache_dir() / "liveresearch_bench"
 
 
 def _load_jsonl(path: Path) -> List[Dict[str, Any]]:

src/openjarvis/evals/datasets/loghub.py

@@ -13,6 +13,7 @@ import re
 from pathlib import Path
 from typing import Any, Dict, Iterable, List, Optional
 
+from openjarvis.core.paths import get_cache_dir
 from openjarvis.evals.core.dataset import DatasetProvider
 from openjarvis.evals.core.types import EvalRecord
 
@@ -64,9 +65,7 @@ class LogHubDataset(DatasetProvider):
                 f"Choose from: {list(_DATASETS.keys())}"
             )
         self._subset = subset
-        self._cache_dir = (
-            Path(cache_dir) if cache_dir else Path.home() / ".cache" / "loghub"
-        )
+        self._cache_dir = Path(cache_dir) if cache_dir else get_cache_dir() / "loghub"
         self._records: List[EvalRecord] = []
 
     def load(

src/openjarvis/evals/datasets/paperarena.py

@@ -15,6 +15,7 @@ import random
 from pathlib import Path
 from typing import Any, Dict, Iterable, List, Optional
 
+from openjarvis.core.paths import get_cache_dir
 from openjarvis.evals.core.dataset import DatasetProvider
 from openjarvis.evals.core.types import EvalRecord
 
@@ -41,7 +42,7 @@ class PaperArenaDataset(DatasetProvider):
         cache_dir: Optional[str] = None,
     ) -> None:
         self._cache_dir = (
-            Path(cache_dir) if cache_dir else Path.home() / ".cache" / "paperarena"
+            Path(cache_dir) if cache_dir else get_cache_dir() / "paperarena"
         )
         self._records: List[EvalRecord] = []
 

src/openjarvis/evals/datasets/pinchbench.py

@@ -18,6 +18,7 @@ from typing import Any, Dict, Iterable, List, Optional
 
 import yaml
 
+from openjarvis.core.paths import get_cache_dir
 from openjarvis.evals.core.dataset import DatasetProvider
 from openjarvis.evals.core.splits import apply_split
 from openjarvis.evals.core.types import EvalRecord
@@ -25,7 +26,7 @@ from openjarvis.evals.core.types import EvalRecord
 LOGGER = logging.getLogger(__name__)
 
 PINCHBENCH_REPO = "https://github.com/pinchbench/skill.git"
-CACHE_DIR = Path.home() / ".cache" / "pinchbench"
+CACHE_DIR = get_cache_dir() / "pinchbench"
 
 
 def _parse_task_markdown(content: str, filename: str = "") -> Dict[str, Any]:

src/openjarvis/evals/datasets/taubench.py

@@ -12,9 +12,9 @@ import logging
 import os
 import subprocess
 import sys
-from pathlib import Path
 from typing import Iterable, List, Optional
 
+from openjarvis.core.paths import get_cache_dir
 from openjarvis.evals.core.dataset import DatasetProvider
 from openjarvis.evals.core.splits import apply_split
 from openjarvis.evals.core.types import EvalRecord
@@ -22,7 +22,7 @@ from openjarvis.evals.core.types import EvalRecord
 LOGGER = logging.getLogger(__name__)
 
 TAU2_REPO = "https://github.com/sierra-research/tau2-bench.git"
-CACHE_DIR = Path.home() / ".cache" / "tau2-bench"
+CACHE_DIR = get_cache_dir() / "tau2-bench"
 
 DOMAINS = ("airline", "retail", "telecom")
 

src/openjarvis/evals/datasets/webchorearena.py

@@ -21,6 +21,7 @@ import subprocess
 from pathlib import Path
 from typing import Any, Dict, Iterable, List, Optional
 
+from openjarvis.core.paths import get_cache_dir
 from openjarvis.evals.core.dataset import DatasetProvider
 from openjarvis.evals.core.types import EvalRecord
 
@@ -56,7 +57,7 @@ class WebChoreArenaDataset(DatasetProvider):
     ) -> None:
         self._subset = subset  # "all", "small", or a site name
         self._cache_dir = (
-            Path(cache_dir) if cache_dir else Path.home() / ".cache" / "webchorearena"
+            Path(cache_dir) if cache_dir else get_cache_dir() / "webchorearena"
         )
         self._headless = headless
         self._records: List[EvalRecord] = []

src/openjarvis/evals/scorers/swebench_harness.py

@@ -49,6 +49,7 @@ import tempfile
 from pathlib import Path
 from typing import Any, Dict, Optional, Tuple
 
+from openjarvis.core.paths import get_cache_dir
 from openjarvis.evals.core.scorer import Scorer
 from openjarvis.evals.core.types import EvalRecord
 
@@ -93,9 +94,7 @@ def _run_subprocess_hard_timeout(
     )
     try:
         stdout, stderr = proc.communicate(timeout=timeout_s)
-        return subprocess.CompletedProcess(
-            cmd, proc.returncode, stdout, stderr
-        )
+        return subprocess.CompletedProcess(cmd, proc.returncode, stdout, stderr)
     except subprocess.TimeoutExpired:
         # Kill the whole group, not just the direct child — Modal harness
         # subprocesses fork workers that would otherwise keep pipes open.
@@ -115,9 +114,7 @@ def _run_subprocess_hard_timeout(
             stdout, stderr = proc.communicate(timeout=10)
         except subprocess.TimeoutExpired:
             stdout, stderr = "", ""
-        raise subprocess.TimeoutExpired(
-            cmd, timeout_s, output=stdout, stderr=stderr
-        )
+        raise subprocess.TimeoutExpired(cmd, timeout_s, output=stdout, stderr=stderr)
 
 
 # ---------- Patch tracking ----------
@@ -208,11 +205,11 @@ def _patch_modal_sandbox_source() -> None:
         # Upstream changed the line — bail rather than apply blindly.
         return
     replacement = (
-        '        # ' + _CGROUP_SOURCE_SENTINEL + '\n'
-        '        try:\n'
+        "        # " + _CGROUP_SOURCE_SENTINEL + "\n"
+        "        try:\n"
         '            self.write_file("/sys/fs/cgroup/cpu/cpu.shares", "2048")\n'
-        '        except FileNotFoundError:\n'
-        '            pass  # cgroup v2 Modal sandbox — path missing is fine\n'
+        "        except FileNotFoundError:\n"
+        "            pass  # cgroup v2 Modal sandbox — path missing is fine\n"
     )
     new_src = src.replace(needle + "\n", replacement, 1)
     try:
@@ -324,22 +321,23 @@ def extract_patch(text: str) -> Optional[str]:
 
 # ---------- Harness invocation ----------
 
+
 def _harness_cache_dir() -> Path:
     """Where the swebench subprocess writes its report JSON + logs/ tree.
 
-    Defaults to ``$OPENJARVIS_HOME/.swebench-cache`` if set, otherwise to a
-    process-shared tempdir. Pin both so we don't pollute the project root.
+    Consolidated under the env-aware OpenJarvis cache root
+    (``<openjarvis-home>/cache/swebench``) so it never pollutes the project
+    root or scatters across ``$HOME``. Honors ``OPENJARVIS_HOME`` /
+    ``XDG_DATA_HOME`` via :func:`openjarvis.core.paths.get_cache_dir`.
     """
-    home = os.environ.get("OPENJARVIS_HOME")
-    if home:
-        cache = Path(home) / ".swebench-cache"
-    else:
-        cache = Path(tempfile.gettempdir()) / "openjarvis-swebench-cache"
+    cache = get_cache_dir() / "swebench"
     cache.mkdir(parents=True, exist_ok=True)
     return cache
 
 
-def _find_report(cache: Path, instance_id: str, run_id: str) -> Optional[Dict[str, Any]]:
+def _find_report(
+    cache: Path, instance_id: str, run_id: str
+) -> Optional[Dict[str, Any]]:
     """Find the harness's report JSON for one instance.
 
     swebench writes ``<model_name_or_path>.<run_id>.json`` inside the
@@ -427,26 +425,40 @@ def _run_harness(
     with tempfile.TemporaryDirectory() as tmp:
         tmp_path = Path(tmp)
         preds_path = tmp_path / "predictions.jsonl"
-        preds_path.write_text(json.dumps({
-            "instance_id": instance_id,
-            "model_name_or_path": "openjarvis-harness",
-            "model_patch": patch,
-        }) + "\n")
+        preds_path.write_text(
+            json.dumps(
+                {
+                    "instance_id": instance_id,
+                    "model_name_or_path": "openjarvis-harness",
+                    "model_patch": patch,
+                }
+            )
+            + "\n"
+        )
 
         cmd = [
-            sys.executable, "-m", "swebench.harness.run_evaluation",
-            "--predictions_path", str(preds_path),
-            "--max_workers", "1",
-            "--run_id", run_id,
-            "--dataset_name", "SWE-bench/SWE-bench_Verified",
-            "--instance_ids", instance_id,
+            sys.executable,
+            "-m",
+            "swebench.harness.run_evaluation",
+            "--predictions_path",
+            str(preds_path),
+            "--max_workers",
+            "1",
+            "--run_id",
+            run_id,
+            "--dataset_name",
+            "SWE-bench/SWE-bench_Verified",
+            "--instance_ids",
+            instance_id,
         ]
         if backend == "modal":
             cmd += ["--modal", "true"]
 
         try:
             proc = _run_subprocess_hard_timeout(
-                cmd, timeout_s=timeout_s, cwd=str(cache),
+                cmd,
+                timeout_s=timeout_s,
+                cwd=str(cache),
             )
         except subprocess.TimeoutExpired as exc:
             # The harness subprocess (and its Modal grandchildren) exceeded
@@ -497,6 +509,7 @@ def _run_harness(
 
 # ---------- Scorer ----------
 
+
 class SWEBenchHarnessScorer(Scorer):
     """SWE-bench Verified scorer that runs the official harness.
 
@@ -516,7 +529,7 @@ class SWEBenchHarnessScorer(Scorer):
         timeout_s: int = 1800,
         cell_name: Optional[str] = None,
         judge_backend: object = None,  # noqa: ARG002 — CLI factory compat
-        judge_model: str = "",         # noqa: ARG002 — CLI factory compat
+        judge_model: str = "",  # noqa: ARG002 — CLI factory compat
     ) -> None:
         self._timeout_s = int(timeout_s)
         # ``cell_name`` namespaces the ``run_id`` so concurrent cells scoring
@@ -538,16 +551,15 @@ class SWEBenchHarnessScorer(Scorer):
         if patch is None:
             return False, {"reason": "no_patch_extracted"}
 
-        instance_id = (
-            record.metadata.get("instance_id")
-            or record.record_id
-            or ""
-        )
+        instance_id = record.metadata.get("instance_id") or record.record_id or ""
         if not instance_id:
             return False, {"reason": "missing_instance_id"}
 
         result = _run_harness(
-            instance_id, patch, self._timeout_s, cell_name=self._cell_name,
+            instance_id,
+            patch,
+            self._timeout_s,
+            cell_name=self._cell_name,
         )
         details = dict(result.get("details", {}))
         details["patch"] = patch

src/openjarvis/evals/skill_benchmark.py

@@ -16,6 +16,8 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+from openjarvis.core.paths import get_config_dir
+
 LOGGER = logging.getLogger(__name__)
 
 CONDITIONS = (
@@ -52,14 +54,12 @@ class SkillBenchmarkConfig:
     seeds: List[int] = field(default_factory=lambda: [42, 43, 44])
     max_samples: Optional[int] = None
     output_dir: Path = field(default_factory=lambda: Path("docs/superpowers/results/"))
-    skills_dir: Path = field(
-        default_factory=lambda: Path("~/.openjarvis/skills/").expanduser()
-    )
+    skills_dir: Path = field(default_factory=lambda: get_config_dir() / "skills")
     overlay_dir_dspy: Path = field(
-        default_factory=lambda: Path("~/.openjarvis/learning/skills-dspy/").expanduser()
+        default_factory=lambda: get_config_dir() / "learning" / "skills-dspy"
     )
     overlay_dir_gepa: Path = field(
-        default_factory=lambda: Path("~/.openjarvis/learning/skills-gepa/").expanduser()
+        default_factory=lambda: get_config_dir() / "learning" / "skills-gepa"
     )
 
 

src/openjarvis/learning/agents/ace_optimizer.py

@@ -30,6 +30,7 @@ from pathlib import Path
 from typing import Any, Dict, List
 
 from openjarvis.core.config import ACEOptimizerConfig
+from openjarvis.core.paths import get_config_dir
 from openjarvis.core.registry import LearningRegistry
 from openjarvis.learning._stubs import AgentLearningPolicy
 
@@ -48,7 +49,7 @@ except ImportError:
 
 
 def _default_save_dir(task_name: str) -> Path:
-    return Path.home() / ".openjarvis" / "learning" / "ace" / task_name
+    return get_config_dir() / "learning" / "ace" / task_name
 
 
 class _TraceDataProcessor:
@@ -87,9 +88,7 @@ class _TraceDataProcessor:
         if not predictions:
             return 0.0
         n_correct = sum(
-            1
-            for p, g in zip(predictions, ground_truths)
-            if cls.answer_is_correct(p, g)
+            1 for p, g in zip(predictions, ground_truths) if cls.answer_is_correct(p, g)
         )
         return n_correct / len(predictions)
 
@@ -149,8 +148,7 @@ class ACEAgentOptimizer:
             return {
                 "status": "skipped",
                 "reason": (
-                    f"only {len(traces)} traces, "
-                    f"min_traces={self.config.min_traces}"
+                    f"only {len(traces)} traces, min_traces={self.config.min_traces}"
                 ),
             }
 
@@ -158,8 +156,7 @@ class ACEAgentOptimizer:
             return {
                 "status": "error",
                 "reason": (
-                    "ace not installed (pip install "
-                    "'openjarvis[learning-ace]')"
+                    "ace not installed (pip install 'openjarvis[learning-ace]')"
                 ),
             }
 

src/openjarvis/learning/agents/skill_optimizer.py

@@ -19,6 +19,7 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+from openjarvis.core.paths import get_config_dir
 from openjarvis.core.types import Trace, TraceStep
 from openjarvis.skills.manager import SkillManager
 from openjarvis.skills.overlay import SkillOverlay, write_overlay
@@ -99,7 +100,7 @@ class SkillOptimizer:
                 pass
             if overlay_dir is None:
                 overlay_dir = Path(
-                    "~/.openjarvis/learning/skills/"
+                    str(get_config_dir() / "learning" / "skills")
                 ).expanduser()
         overlay_dir = Path(overlay_dir).expanduser()
 

src/openjarvis/learning/spec_search/storage/paths.py

@@ -11,14 +11,21 @@ writing artifacts into the working tree.
 
 from __future__ import annotations
 
-import os
 from pathlib import Path
 
+from openjarvis.core.paths import ConfigurationError, get_config_dir
 from openjarvis.security.file_utils import secure_mkdir
 
-
-class ConfigurationError(RuntimeError):
-    """Raised when path configuration would violate isolation guarantees."""
+# ``ConfigurationError`` is re-exported from ``openjarvis.core.paths`` (it used
+# to be defined here). Spec search now resolves the home dir through the unified
+# core resolver, which raises the same exception type on a source-tree path, so
+# we alias rather than redefine to keep ``except ConfigurationError`` callers and
+# existing tests working.
+__all__ = [
+    "ConfigurationError",
+    "ensure_spec_search_dirs",
+    "resolve_spec_search_root",
+]
 
 
 def _find_source_root() -> Path | None:
@@ -42,11 +49,12 @@ def _find_source_root() -> Path | None:
 
 
 def _resolve_openjarvis_home() -> Path:
-    """Resolve the OPENJARVIS_HOME directory (env var or default)."""
-    env = os.environ.get("OPENJARVIS_HOME")
-    if env:
-        return Path(env).expanduser().resolve()
-    return (Path.home() / ".openjarvis").resolve()
+    """Resolve the OpenJarvis home directory via the unified core resolver.
+
+    Delegates to ``get_config_dir`` so spec-search honors the same env-aware
+    resolution (OPENJARVIS_HOME and XDG) as the rest of the framework.
+    """
+    return get_config_dir()
 
 
 def resolve_spec_search_root() -> Path:

src/openjarvis/mining/_constants.py

@@ -9,7 +9,7 @@ section 7.3 for the rev-bump workflow.
 
 from __future__ import annotations
 
-from pathlib import Path
+from openjarvis.core.paths import get_config_dir
 
 PEARL_REPO = "https://github.com/pearl-research-labs/pearl.git"
 # TODO at implementation time: replace with the specific commit/tag verified
@@ -47,9 +47,9 @@ DEFAULT_PEARLD_RPC_URL = "http://localhost:44107"
 MIN_FREE_DISK_GB = 200
 
 # Runtime sidecar location (single-session assumption — see spec §8.8).
-RUNTIME_DIR = Path.home() / ".openjarvis" / "runtime"
+RUNTIME_DIR = get_config_dir() / "runtime"
 SIDECAR_PATH = RUNTIME_DIR / "mining.json"
 SIDECAR_LOCK_PATH = RUNTIME_DIR / "mining.lock"
 
 # Pearl source cache for build-from-pin path (see spec §7.2).
-PEARL_CACHE_DIR = Path.home() / ".openjarvis" / "cache" / "pearl"
+PEARL_CACHE_DIR = get_config_dir() / "cache" / "pearl"

src/openjarvis/mining/cpu_pearl.py

@@ -17,6 +17,7 @@ import urllib.request
 from pathlib import Path
 
 from openjarvis.core.config import HardwareInfo
+from openjarvis.core.paths import get_config_dir
 from openjarvis.core.registry import MinerRegistry
 
 from . import _install
@@ -47,7 +48,7 @@ def _sidecar_path() -> Path:
 
 def _log_dir() -> Path:
     """Return the logs directory. Override in tests."""
-    return Path.home() / ".openjarvis" / "logs" / "mining"
+    return get_config_dir() / "logs" / "mining"
 
 
 def _parse_gateway_metrics(text: str, *, provider_id: str) -> MiningStats:

src/openjarvis/prompt/builder.py

@@ -5,6 +5,7 @@ from pathlib import Path
 from typing import List, Literal, Optional, Tuple
 
 from openjarvis.core.config import MemoryFilesConfig, SystemPromptConfig
+from openjarvis.core.paths import get_config_dir
 
 PromptCacheSegment = Literal["frozen_prefix", "dynamic_suffix"]
 
@@ -279,7 +280,7 @@ class SystemPromptBuilder:
                 f"Invalid persona name {name!r}: must be a simple "
                 "identifier (no path separators or '..')."
             )
-        base = Path.home() / ".openjarvis" / "personas" / name
+        base = get_config_dir() / "personas" / name
         return MemoryFilesConfig(
             soul_path=str(base / "SOUL.md"),
             memory_path=str(base / "MEMORY.md"),

src/openjarvis/recipes/loader.py

@@ -19,13 +19,14 @@ try:
 except ModuleNotFoundError:
     import tomli as tomllib  # type: ignore[no-redef]
 
+from openjarvis.core.paths import get_config_dir
 
 # Built-in recipes directory (package data)
 _PROJECT_RECIPES_DIR = Path(__file__).resolve().parent / "data"
 _PROJECT_OPERATORS_DIR = _PROJECT_RECIPES_DIR / "operators"
 # User-level directories
-_USER_RECIPES_DIR = Path.home() / ".openjarvis" / "recipes"
-_USER_OPERATORS_DIR = Path.home() / ".openjarvis" / "operators"
+_USER_RECIPES_DIR = get_config_dir() / "recipes"
+_USER_OPERATORS_DIR = get_config_dir() / "operators"
 
 
 @dataclass(slots=True)

src/openjarvis/server/agent_manager_routes.py

@@ -2095,10 +2095,13 @@ def create_agent_manager_router(
             raise HTTPException(status_code=404, detail="Agent not found")
         try:
             from openjarvis.core.config import load_config
+            from openjarvis.core.paths import get_config_dir
             from openjarvis.traces.store import TraceStore
 
             config = load_config()
-            store = TraceStore(config.traces.db_path or "~/.openjarvis/traces.db")
+            store = TraceStore(
+                config.traces.db_path or str(get_config_dir() / "traces.db")
+            )
             traces = store.list_traces(agent=agent_id, limit=limit)
             return {
                 "traces": [
@@ -2120,10 +2123,13 @@ def create_agent_manager_router(
     def get_trace(agent_id: str, trace_id: str):
         try:
             from openjarvis.core.config import load_config
+            from openjarvis.core.paths import get_config_dir
             from openjarvis.traces.store import TraceStore
 
             config = load_config()
-            store = TraceStore(config.traces.db_path or "~/.openjarvis/traces.db")
+            store = TraceStore(
+                config.traces.db_path or str(get_config_dir() / "traces.db")
+            )
             trace = store.get(trace_id)
             if trace is None:
                 raise HTTPException(status_code=404, detail="Trace not found")

src/openjarvis/server/cloud_router.py

@@ -10,18 +10,18 @@ from __future__ import annotations
 import json
 import os
 from collections.abc import AsyncIterator
-from pathlib import Path
 from typing import Any, Sequence
 
 import httpx
 
+from openjarvis.core.paths import get_config_dir
 from openjarvis.core.types import Message
 
 # ---------------------------------------------------------------------------
 # Key / provider detection
 # ---------------------------------------------------------------------------
 
-_CLOUD_ENV_FILE = Path.home() / ".openjarvis" / "cloud-keys.env"
+_CLOUD_ENV_FILE = get_config_dir() / "cloud-keys.env"
 
 _OPENAI_PREFIXES = ("gpt-", "o1-", "o3-", "o4-", "chatgpt-")
 _ANTHROPIC_PREFIXES = ("claude-",)

src/openjarvis/server/routes.py

@@ -9,6 +9,7 @@ from typing import Any
 from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import StreamingResponse
 
+from openjarvis.core.paths import get_config_dir
 from openjarvis.core.types import Message, Role
 from openjarvis.server.models import (
     ChatCompletionChunk,
@@ -821,10 +822,9 @@ async def reload_cloud_engine(request: Request):
     key so that cloud models become available without a full app restart.
     """
     import os
-    from pathlib import Path
 
     # Re-read ~/.openjarvis/cloud-keys.env and update the running process env.
-    keys_path = Path.home() / ".openjarvis" / "cloud-keys.env"
+    keys_path = get_config_dir() / "cloud-keys.env"
     if keys_path.exists():
         for raw_line in keys_path.read_text().splitlines():
             line = raw_line.strip()

src/openjarvis/server/session_store.py

@@ -8,6 +8,8 @@ import sqlite3
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+from openjarvis.core.paths import get_config_dir
+
 logger = logging.getLogger(__name__)
 
 _MAX_HISTORY_TURNS = 20
@@ -22,7 +24,7 @@ class SessionStore:
 
     def __init__(self, db_path: str = "") -> None:
         if not db_path:
-            db_path = str(Path.home() / ".openjarvis" / "sessions.db")
+            db_path = str(get_config_dir() / "sessions.db")
         from openjarvis.security.file_utils import secure_create
 
         secure_create(Path(db_path))

src/openjarvis/skills/importer.py

@@ -23,6 +23,7 @@ from typing import List
 
 import yaml
 
+from openjarvis.core.paths import get_config_dir
 from openjarvis.skills.parser import SkillParser
 from openjarvis.skills.sources.base import ResolvedSkill
 from openjarvis.skills.tool_translator import ToolTranslator
@@ -56,7 +57,7 @@ class SkillImporter:
         self._parser = parser
         self._translator = tool_translator
         if target_root is None:
-            target_root = Path("~/.openjarvis/skills/").expanduser()
+            target_root = get_config_dir() / "skills"
         self._target_root = Path(target_root)
 
     def import_skill(

src/openjarvis/skills/manager.py

@@ -7,6 +7,7 @@ from pathlib import Path
 from typing import Any, Dict, List, Optional
 
 from openjarvis.core.events import EventBus
+from openjarvis.core.paths import get_config_dir
 from openjarvis.skills.dependency import validate_dependencies
 from openjarvis.skills.executor import SkillExecutor, SkillResult
 from openjarvis.skills.loader import discover_skills
@@ -54,7 +55,7 @@ class SkillManager:
             except Exception:
                 pass
             if overlay_dir is None:
-                overlay_dir = Path("~/.openjarvis/learning/skills/").expanduser()
+                overlay_dir = get_config_dir() / "learning" / "skills"
         self._overlay_dir = Path(overlay_dir).expanduser()
 
     # ------------------------------------------------------------------
@@ -262,7 +263,7 @@ class SkillManager:
         discovered = discovery.analyze_traces(traces)
 
         if output_dir is None:
-            output_dir = Path("~/.openjarvis/skills/discovered/").expanduser()
+            output_dir = get_config_dir() / "skills" / "discovered"
         output_dir = Path(output_dir).expanduser()
         output_dir.mkdir(parents=True, exist_ok=True)
 
@@ -375,7 +376,7 @@ class SkillManager:
         manifest's ``name`` field equals ``name``.
         """
         if roots is None:
-            roots = [Path("~/.openjarvis/skills/").expanduser(), Path("./skills")]
+            roots = [get_config_dir() / "skills", Path("./skills")]
 
         matches: List[Path] = []
         for root in roots:

src/openjarvis/skills/sources/hermes.py

@@ -14,6 +14,7 @@ from typing import List
 
 import yaml
 
+from openjarvis.core.paths import get_config_dir
 from openjarvis.skills.sources.base import ResolvedSkill, SourceResolver
 
 LOGGER = logging.getLogger(__name__)
@@ -28,7 +29,7 @@ class HermesResolver(SourceResolver):
 
     def __init__(self, cache_root: Path | None = None) -> None:
         if cache_root is None:
-            cache_root = Path("~/.openjarvis/skill-cache/hermes/").expanduser()
+            cache_root = get_config_dir() / "skill-cache" / "hermes"
         self._cache_root = Path(cache_root)
 
     def cache_dir(self) -> Path:

src/openjarvis/skills/sources/openclaw.py

@@ -15,6 +15,7 @@ from typing import List
 
 import yaml
 
+from openjarvis.core.paths import get_config_dir
 from openjarvis.skills.sources.base import ResolvedSkill, SourceResolver
 
 LOGGER = logging.getLogger(__name__)
@@ -29,7 +30,7 @@ class OpenClawResolver(SourceResolver):
 
     def __init__(self, cache_root: Path | None = None) -> None:
         if cache_root is None:
-            cache_root = Path("~/.openjarvis/skill-cache/openclaw/").expanduser()
+            cache_root = get_config_dir() / "skill-cache" / "openclaw"
         self._cache_root = Path(cache_root)
 
     def cache_dir(self) -> Path:

src/openjarvis/system/builder.py

@@ -7,6 +7,7 @@ from typing import Any, List, Optional
 
 from openjarvis.core.config import JarvisConfig, load_config
 from openjarvis.core.events import EventBus, get_event_bus
+from openjarvis.core.paths import get_config_dir
 from openjarvis.engine._stubs import InferenceEngine
 from openjarvis.system.core import JarvisSystem
 from openjarvis.tools._stubs import BaseTool, ToolExecutor
@@ -232,12 +233,10 @@ class SystemBuilder:
         agent_manager = None
         if config.agent_manager.enabled:
             try:
-                from pathlib import Path
-
                 from openjarvis.agents.manager import AgentManager
 
                 am_db = config.agent_manager.db_path or str(
-                    Path("~/.openjarvis/agents.db").expanduser()
+                    get_config_dir() / "agents.db"
                 )
                 agent_manager = AgentManager(db_path=am_db)
             except Exception as exc:

src/openjarvis/templates/agent_templates.py

@@ -11,6 +11,8 @@ try:
 except ModuleNotFoundError:
     import tomli as tomllib  # type: ignore[no-redef]
 
+from openjarvis.core.paths import get_config_dir
+
 
 @dataclass(slots=True)
 class AgentTemplate:
@@ -72,7 +74,7 @@ def _builtin_templates_dir() -> Path:
 
 def _user_templates_dir() -> Path:
     """Return the path to user-defined templates (~/.openjarvis/templates/agents/)."""
-    return Path.home() / ".openjarvis" / "templates" / "agents"
+    return get_config_dir() / "templates" / "agents"
 
 
 def discover_templates(

src/openjarvis/tools/approval_store.py

@@ -19,6 +19,8 @@ from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+from openjarvis.core.paths import get_config_dir
+
 # ---------------------------------------------------------------------------
 # Decision constants
 # ---------------------------------------------------------------------------
@@ -147,7 +149,7 @@ class ApprovalStore:
 
     def __init__(self, db_path: str = "") -> None:
         if not db_path:
-            db_path = str(Path.home() / ".openjarvis" / "approvals.db")
+            db_path = str(get_config_dir() / "approvals.db")
         self._db_path = db_path
         Path(db_path).parent.mkdir(parents=True, exist_ok=True)
         self._conn = sqlite3.connect(db_path, check_same_thread=False)

src/openjarvis/tools/description_loader.py

@@ -16,10 +16,10 @@ The TOML file format (written by the applier) is::
 from __future__ import annotations
 
 import logging
-import os
-from pathlib import Path
 from typing import Dict, Optional
 
+from openjarvis.core.paths import get_config_dir
+
 logger = logging.getLogger(__name__)
 
 _cache: Optional[Dict[str, str]] = None
@@ -27,7 +27,7 @@ _cache: Optional[Dict[str, str]] = None
 
 def _load_overrides() -> Dict[str, str]:
     """Parse descriptions.toml and return {tool_name: description}."""
-    home = Path(os.environ.get("OPENJARVIS_HOME", "~/.openjarvis")).expanduser()
+    home = get_config_dir()
     desc_path = home / "tools" / "descriptions.toml"
     if not desc_path.exists():
         return {}

src/openjarvis/tools/memory_manage.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 from pathlib import Path
 from typing import Any
 
+from openjarvis.core.paths import get_config_dir
 from openjarvis.core.registry import ToolRegistry
 from openjarvis.core.types import ToolResult
 from openjarvis.tools._stubs import BaseTool, ToolSpec
@@ -14,7 +15,9 @@ from openjarvis.tools._stubs import BaseTool, ToolSpec
 class MemoryManageTool(BaseTool):
     """Manage persistent agent memory (MEMORY.md)."""
 
-    def __init__(self, memory_path: Path | str = "~/.openjarvis/MEMORY.md") -> None:
+    def __init__(self, memory_path: Path | str | None = None) -> None:
+        if memory_path is None:
+            memory_path = get_config_dir() / "MEMORY.md"
         self._memory_path = Path(memory_path).expanduser()
 
     @property

src/openjarvis/tools/skill_manage.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 from pathlib import Path
 from typing import Any, List
 
+from openjarvis.core.paths import get_config_dir
 from openjarvis.core.registry import ToolRegistry
 from openjarvis.core.types import ToolResult
 from openjarvis.tools._stubs import BaseTool, ToolSpec
@@ -14,7 +15,9 @@ from openjarvis.tools._stubs import BaseTool, ToolSpec
 class SkillManageTool(BaseTool):
     """Manage agent-authored procedural skills."""
 
-    def __init__(self, skills_dir: Path | str = "~/.openjarvis/skills/") -> None:
+    def __init__(self, skills_dir: Path | str | None = None) -> None:
+        if skills_dir is None:
+            skills_dir = get_config_dir() / "skills"
         self._skills_dir = Path(skills_dir).expanduser()
 
     @property

src/openjarvis/tools/user_profile_manage.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 from pathlib import Path
 from typing import Any
 
+from openjarvis.core.paths import get_config_dir
 from openjarvis.core.registry import ToolRegistry
 from openjarvis.core.types import ToolResult
 from openjarvis.tools._stubs import BaseTool, ToolSpec
@@ -14,7 +15,9 @@ from openjarvis.tools._stubs import BaseTool, ToolSpec
 class UserProfileManageTool(BaseTool):
     """Manage persistent user profile (USER.md)."""
 
-    def __init__(self, user_path: Path | str = "~/.openjarvis/USER.md") -> None:
+    def __init__(self, user_path: Path | str | None = None) -> None:
+        if user_path is None:
+            user_path = get_config_dir() / "USER.md"
         self._user_path = Path(user_path).expanduser()
 
     @property

src/openjarvis/tools/web_search.py

@@ -165,7 +165,10 @@ class WebSearchTool(BaseTool):
 
             client = TavilyClient(api_key=self._api_key)
             response = client.search(
-                query, max_results=max_results, search_depth="advanced"
+                query,
+                max_results=max_results,
+                search_depth="advanced",
+                include_usage=True,
             )
             results = response.get("results", [])
             formatted_parts = []
@@ -182,7 +185,11 @@ class WebSearchTool(BaseTool):
                 tool_name="web_search",
                 content=formatted or "No results found.",
                 success=True,
-                metadata={"num_results": len(results), "engine": "tavily"},
+                metadata={
+                    "num_results": len(results),
+                    "engine": "tavily",
+                    "credits": (response.get("usage") or {}).get("credits"),
+                },
             )
         except Exception as exc:
             logger.debug(

tests/core/test_config_paths.py

@@ -0,0 +1,176 @@
+"""Tests for the env-aware OpenJarvis home-directory resolver (issue #462).
+
+Covers the single-root consolidation: ``$OPENJARVIS_HOME`` >
+``$XDG_DATA_HOME/openjarvis`` > ``~/.openjarvis``, backward compatibility
+(no env => exactly ``~/.openjarvis``), and the source-tree rejection guard.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from openjarvis.core import paths
+
+
+def _clear_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Remove every env var that influences home resolution."""
+    for var in (
+        "OPENJARVIS_HOME",
+        "XDG_DATA_HOME",
+        "XDG_CONFIG_HOME",
+        "XDG_CACHE_HOME",
+    ):
+        monkeypatch.delenv(var, raising=False)
+
+
+class TestGetConfigDir:
+    """Precedence and backward compatibility of get_config_dir()."""
+
+    def test_default_when_unset_is_legacy_dir(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # Backward-compat: with nothing set, the resolved dir is exactly the
+        # historical ~/.openjarvis so existing installs are untouched.
+        _clear_env(monkeypatch)
+        assert paths.get_config_dir() == (Path.home() / ".openjarvis").resolve()
+
+    def test_respects_openjarvis_home(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _clear_env(monkeypatch)
+        custom = tmp_path / "oj"
+        monkeypatch.setenv("OPENJARVIS_HOME", str(custom))
+        assert paths.get_config_dir() == custom.resolve()
+
+    def test_respects_xdg_data_home(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _clear_env(monkeypatch)
+        monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path))
+        # Single nested 'openjarvis' dir under XDG_DATA_HOME.
+        assert paths.get_config_dir() == (tmp_path / "openjarvis").resolve()
+
+    def test_openjarvis_home_wins_over_xdg(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _clear_env(monkeypatch)
+        oj = tmp_path / "oj_wins"
+        monkeypatch.setenv("OPENJARVIS_HOME", str(oj))
+        monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "xdg_loses"))
+        assert paths.get_config_dir() == oj.resolve()
+
+    def test_expands_user_in_openjarvis_home(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _clear_env(monkeypatch)
+        monkeypatch.setenv("OPENJARVIS_HOME", "~/relocated-oj")
+        assert paths.get_config_dir() == (Path.home() / "relocated-oj").resolve()
+
+    def test_returns_absolute_path(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _clear_env(monkeypatch)
+        monkeypatch.setenv("OPENJARVIS_HOME", str(tmp_path / "rel"))
+        assert paths.get_config_dir().is_absolute()
+
+
+class TestDerivedDirs:
+    """config_path / data_dir / cache_dir all hang off the single root."""
+
+    def test_config_path(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        _clear_env(monkeypatch)
+        monkeypatch.setenv("OPENJARVIS_HOME", str(tmp_path / "oj"))
+        assert paths.get_config_path() == (tmp_path / "oj" / "config.toml").resolve()
+
+    def test_data_dir_equals_config_dir(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _clear_env(monkeypatch)
+        monkeypatch.setenv("OPENJARVIS_HOME", str(tmp_path / "oj"))
+        assert paths.get_data_dir() == paths.get_config_dir()
+
+    def test_cache_dir_is_nested_cache(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _clear_env(monkeypatch)
+        monkeypatch.setenv("OPENJARVIS_HOME", str(tmp_path / "oj"))
+        assert paths.get_cache_dir() == (tmp_path / "oj" / "cache").resolve()
+
+    def test_cache_dir_under_xdg(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _clear_env(monkeypatch)
+        monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path))
+        assert paths.get_cache_dir() == (tmp_path / "openjarvis" / "cache").resolve()
+
+
+class TestSourceTreeRejection:
+    """A home pointing inside the repo must fail loudly (REVIEW.md)."""
+
+    def test_rejects_path_inside_source_tree(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _clear_env(monkeypatch)
+        source_root = paths._find_source_root()
+        assert source_root is not None  # We must be running inside the repo.
+        monkeypatch.setenv("OPENJARVIS_HOME", str(source_root / "junk_dir"))
+        with pytest.raises(paths.ConfigurationError, match="inside the source tree"):
+            paths.get_config_dir()
+
+
+class TestLegacyConstantsHonorEnv:
+    """The legacy DEFAULT_CONFIG_* names route through the env-aware resolver.
+
+    This is the exact split-brain bug from #462: the constant used to ignore
+    OPENJARVIS_HOME entirely. The constant is resolved once at import (the
+    install-script model, where the env is set before the process starts), and
+    every instance-level default goes through ``get_config_dir()`` so it honors
+    the override. ``DEFAULT_CONFIG_DIR`` stays a real attribute so existing
+    tests can ``monkeypatch.setattr`` it.
+    """
+
+    def test_constant_matches_resolver_at_import(self) -> None:
+        from openjarvis.core import config
+
+        # The constant is the import-time resolution of the same function.
+        assert config.DEFAULT_CONFIG_DIR == paths.get_config_dir()
+        assert config.DEFAULT_CONFIG_PATH == paths.get_config_path()
+
+    def test_constant_is_a_real_settable_attribute(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # Install/CLI tests monkeypatch this attribute directly; it must be a
+        # real module attribute (not __getattr__-only) for setattr/undo to work.
+        from openjarvis.core import config
+
+        monkeypatch.setattr(config, "DEFAULT_CONFIG_DIR", tmp_path / "patched")
+        assert config.DEFAULT_CONFIG_DIR == tmp_path / "patched"
+
+    def test_dataclass_defaults_reflect_env(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # Config dataclass field defaults must resolve under the override at
+        # instantiation time, not freeze ~/.openjarvis at import.
+        _clear_env(monkeypatch)
+        from openjarvis.core.config import SessionConfig, StorageConfig
+
+        monkeypatch.setenv("OPENJARVIS_HOME", str(tmp_path / "oj"))
+        root = (tmp_path / "oj").resolve()
+        assert StorageConfig().db_path == str(root / "memory.db")
+        assert SessionConfig().db_path == str(root / "sessions.db")
+
+    def test_downstream_consumer_honors_env(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # End-to-end: a non-config subsystem (credentials) resolves under the
+        # custom root, proving the override is no longer split-brain.
+        _clear_env(monkeypatch)
+        from openjarvis.core import credentials
+
+        monkeypatch.setenv("OPENJARVIS_HOME", str(tmp_path / "oj"))
+        assert (
+            credentials._default_path()
+            == (tmp_path / "oj" / "credentials.toml").resolve()
+        )

tests/core/test_skills_learning_config.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from openjarvis.core.config import LearningConfig, SkillsLearningConfig
+from openjarvis.core.paths import get_config_dir
 
 
 class TestSkillsLearningConfig:
@@ -12,7 +13,9 @@ class TestSkillsLearningConfig:
         assert cfg.optimizer == "dspy"
         assert cfg.min_traces_per_skill == 20
         assert cfg.optimization_interval_seconds == 86400
-        assert cfg.overlay_dir == "~/.openjarvis/learning/skills/"
+        # overlay_dir now resolves under the env-aware OpenJarvis root (#462),
+        # defaulting to <home>/learning/skills instead of the old literal.
+        assert cfg.overlay_dir == str(get_config_dir() / "learning" / "skills")
 
     def test_can_be_constructed_with_all_fields(self):
         cfg = SkillsLearningConfig(

tests/tools/test_web_search.py

@@ -171,7 +171,7 @@ class TestWebSearchTool:
         tool = WebSearchTool(api_key="test-key", max_results=3)
         tool.execute(query="test", max_results=7)
         mock_client.search.assert_called_once_with(
-            "test", max_results=7, search_depth="advanced"
+            "test", max_results=7, search_depth="advanced", include_usage=True
         )
 
     def test_to_openai_function(self):