fix: reuse parent system prompt for background review to preserve cache bytes

agent/auxiliary_client.py

@@ -88,6 +88,9 @@ class _OpenAIProxy:
     __slots__ = ()
 
     def __call__(self, *args, **kwargs):
+        if "base_url" in kwargs:
+            kwargs = dict(kwargs)
+            kwargs.update(_build_openai_client_kwargs(kwargs.get("base_url"), kwargs))
         return _load_openai_cls()(*args, **kwargs)
 
     def __instancecheck__(self, obj):
@@ -115,6 +118,37 @@ def _safe_isinstance(obj: Any, maybe_type: Any) -> bool:
         return False
 
 
+def _is_localhost_base_url(base_url: Optional[str]) -> bool:
+    try:
+        return base_url_hostname(base_url or "").lower().rstrip(".") in {"localhost", "127.0.0.1", "::1"}
+    except Exception:
+        return False
+
+
+def _build_openai_client_kwargs(base_url: Optional[str], extra: Dict[str, Any] = None) -> Dict[str, Any]:
+    """Build OpenAI kwargs and disable env-derived proxy config for localhost."""
+    kwargs: Dict[str, Any] = dict(extra or {})
+    try:
+        if _is_localhost_base_url(base_url) and "http_client" not in kwargs:
+            import httpx
+            kwargs["http_client"] = httpx.Client(trust_env=False)
+    except Exception:
+        pass
+    return kwargs
+
+
+def _build_async_openai_client_kwargs(base_url: Optional[str], extra: Dict[str, Any] = None) -> Dict[str, Any]:
+    """Build AsyncOpenAI kwargs and disable env-derived proxy config for localhost."""
+    kwargs: Dict[str, Any] = dict(extra or {})
+    try:
+        if _is_localhost_base_url(base_url) and "http_client" not in kwargs:
+            import httpx
+            kwargs["http_client"] = httpx.AsyncClient(trust_env=False)
+    except Exception:
+        pass
+    return kwargs
+
+
 def _extract_url_query_params(url: str):
     """Extract query params from URL, return (clean_url, default_query dict or None)."""
     parsed = urlparse(url)
@@ -2608,11 +2642,12 @@ def _to_async_client(sync_client, model: str, is_vision: bool = False):
     except ImportError:
         pass
 
+    sync_base_url = str(sync_client.base_url)
     async_kwargs = {
         "api_key": sync_client.api_key,
-        "base_url": str(sync_client.base_url),
+        "base_url": sync_base_url,
     }
-    sync_base_url = str(sync_client.base_url)
+    async_kwargs = _build_async_openai_client_kwargs(sync_base_url, async_kwargs)
     if base_url_host_matches(sync_base_url, "openrouter.ai"):
         async_kwargs["default_headers"] = build_or_headers()
     elif base_url_host_matches(sync_base_url, "api.githubcopilot.com"):

cron/jobs.py

@@ -37,15 +37,55 @@
 HERMES_DIR = get_hermes_home().resolve()
 CRON_DIR = HERMES_DIR / "cron"
 JOBS_FILE = CRON_DIR / "jobs.json"
+DEFAULT_HERMES_DIR = HERMES_DIR
+DEFAULT_CRON_DIR = CRON_DIR
+DEFAULT_JOBS_FILE = JOBS_FILE
 
 # In-process lock protecting load_jobs→modify→save_jobs cycles.
 # Required when tick() runs jobs in parallel threads — without this,
 # concurrent mark_job_run / advance_next_run calls can clobber each other.
 _jobs_file_lock = threading.Lock()
 OUTPUT_DIR = CRON_DIR / "output"
+DEFAULT_OUTPUT_DIR = OUTPUT_DIR
 ONESHOT_GRACE_SECONDS = 120
 
 
+def _resolve_runtime_paths() -> tuple[Path, Path, Path]:
+    """Resolve cron directories at runtime.
+
+    Keep paths aligned with the current HERMES_HOME while still honoring
+    test-only explicit monkeypatches of CRON_DIR/JOBS_FILE/OUTPUT_DIR.
+    """
+    # Derive from current HERMES_HOME unless the module attribute has been
+    # explicitly overridden.
+    resolved_home = get_hermes_home().resolve()
+    cron_dir = Path(CRON_DIR)
+    if CRON_DIR == DEFAULT_CRON_DIR:
+        cron_dir = resolved_home / "cron"
+
+    jobs_file = Path(JOBS_FILE)
+    if JOBS_FILE == DEFAULT_JOBS_FILE:
+        jobs_file = cron_dir / "jobs.json"
+
+    output_dir = Path(OUTPUT_DIR)
+    if OUTPUT_DIR == DEFAULT_OUTPUT_DIR:
+        output_dir = cron_dir / "output"
+
+    return cron_dir, jobs_file, output_dir
+
+
+def _get_cron_dir() -> Path:
+    return _resolve_runtime_paths()[0]
+
+
+def _get_jobs_file() -> Path:
+    return _resolve_runtime_paths()[1]
+
+
+def _get_output_dir() -> Path:
+    return _resolve_runtime_paths()[2]
+
+
 def _normalize_skill_list(skill: Optional[str] = None, skills: Optional[Any] = None) -> List[str]:
     """Normalize legacy/single-skill and multi-skill inputs into a unique ordered list."""
     if skills is None:
@@ -150,10 +190,12 @@ def _secure_file(path: Path):
 
 def ensure_dirs():
     """Ensure cron directories exist with secure permissions."""
-    CRON_DIR.mkdir(parents=True, exist_ok=True)
-    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
-    _secure_dir(CRON_DIR)
-    _secure_dir(OUTPUT_DIR)
+    cron_dir = _get_cron_dir()
+    output_dir = _get_output_dir()
+    cron_dir.mkdir(parents=True, exist_ok=True)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    _secure_dir(cron_dir)
+    _secure_dir(output_dir)
 
 
 # =============================================================================
@@ -401,17 +443,18 @@ def compute_next_run(schedule: Dict[str, Any], last_run_at: Optional[str] = None
 def load_jobs() -> List[Dict[str, Any]]:
     """Load all jobs from storage."""
     ensure_dirs()
-    if not JOBS_FILE.exists():
+    jobs_file = _get_jobs_file()
+    if not jobs_file.exists():
         return []
 
     try:
-        with open(JOBS_FILE, 'r', encoding='utf-8') as f:
+        with open(jobs_file, 'r', encoding='utf-8') as f:
             data = json.load(f)
             return data.get("jobs", [])
     except json.JSONDecodeError:
         # Retry with strict=False to handle bare control chars in string values
         try:
-            with open(JOBS_FILE, 'r', encoding='utf-8') as f:
+            with open(jobs_file, 'r', encoding='utf-8') as f:
                 data = json.loads(f.read(), strict=False)
                 jobs = data.get("jobs", [])
                 if jobs:
@@ -430,14 +473,15 @@ def load_jobs() -> List[Dict[str, Any]]:
 def save_jobs(jobs: List[Dict[str, Any]]):
     """Save all jobs to storage."""
     ensure_dirs()
-    fd, tmp_path = tempfile.mkstemp(dir=str(JOBS_FILE.parent), suffix='.tmp', prefix='.jobs_')
+    jobs_file = _get_jobs_file()
+    fd, tmp_path = tempfile.mkstemp(dir=str(jobs_file.parent), suffix='.tmp', prefix='.jobs_')
     try:
         with os.fdopen(fd, 'w', encoding='utf-8') as f:
             json.dump({"jobs": jobs, "updated_at": _hermes_now().isoformat()}, f, indent=2)
             f.flush()
             os.fsync(f.fileno())
-        atomic_replace(tmp_path, JOBS_FILE)
-        _secure_file(JOBS_FILE)
+        atomic_replace(tmp_path, jobs_file)
+        _secure_file(jobs_file)
     except BaseException:
         try:
             os.unlink(tmp_path)
@@ -758,7 +802,7 @@ def remove_job(job_id: str) -> bool:
     if len(jobs) < original_len:
         save_jobs(jobs)
         # Clean up output directory to prevent orphaned dirs accumulating
-        job_output_dir = OUTPUT_DIR / job_id
+        job_output_dir = _get_output_dir() / job_id
         if job_output_dir.exists():
             shutil.rmtree(job_output_dir)
         return True
@@ -971,8 +1015,10 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
 
 def save_job_output(job_id: str, output: str):
     """Save job output to file."""
+    output_dir = _get_output_dir()
+
     ensure_dirs()
-    job_output_dir = OUTPUT_DIR / job_id
+    job_output_dir = output_dir / job_id
     job_output_dir.mkdir(parents=True, exist_ok=True)
     _secure_dir(job_output_dir)
 

cron/scheduler.py

@@ -879,7 +879,7 @@ def _build_job_prompt(job: dict, prerun_script: Optional[tuple] = None) -> str:
     # Inject output from referenced cron jobs as context.
     context_from = job.get("context_from")
     if context_from:
-        from cron.jobs import OUTPUT_DIR
+        from cron.jobs import _get_output_dir
         if isinstance(context_from, str):
             context_from = [context_from]
         for source_job_id in context_from:
@@ -888,7 +888,7 @@ def _build_job_prompt(job: dict, prerun_script: Optional[tuple] = None) -> str:
                 logger.warning("context_from: skipping invalid job_id %r", source_job_id)
                 continue
             try:
-                job_output_dir = OUTPUT_DIR / source_job_id
+                job_output_dir = _get_output_dir() / source_job_id
                 if not job_output_dir.exists():
                     continue  # silent skip — no output yet
                 output_files = sorted(

run_agent.py

@@ -1184,6 +1184,7 @@ def __init__(
         checkpoint_max_total_size_mb: int = 500,
         checkpoint_max_file_size_mb: int = 10,
         pass_session_id: bool = False,
+        cached_system_prompt: Optional[str] = None,
     ):
         """
         Initialize the AI Agent.
@@ -1212,6 +1213,7 @@ def __init__(
                 openrouter/pareto-code router. Only applied when model == "openrouter/pareto-code".
                 None or empty = let OpenRouter pick the strongest available coder.
             session_id (str): Pre-generated session ID for logging (optional, auto-generated if not provided)
+            cached_system_prompt (str): Prebuilt system prompt that should be reused verbatim.
             tool_progress_callback (callable): Callback function(tool_name, args_preview) for progress notifications
             clarify_callback (callable): Callback function(question, choices) -> str for interactive user questions.
                 Provided by the platform layer (CLI or gateway). If None, the clarify tool returns an error.
@@ -1918,7 +1920,7 @@ def __init__(
         self._memory_write_context = "foreground"
 
         # Cached system prompt -- built once per session, only rebuilt on compression
-        self._cached_system_prompt: Optional[str] = None
+        self._cached_system_prompt: Optional[str] = cached_system_prompt
 
         # Filesystem checkpoint manager (transparent — not a tool)
         from tools.checkpoint_manager import CheckpointManager
@@ -4267,6 +4269,7 @@ def _bg_review_auto_deny(command, description, **kwargs):
                     # reconstruct auth from scratch -- producing the spurious
                     # "No LLM provider configured" warning at end of turn.
                     _parent_runtime = self._current_main_runtime()
+                    review_system_prompt = self._cached_system_prompt or self._build_system_prompt()
                     review_agent = AIAgent(
                         model=self.model,
                         max_iterations=16,
@@ -4278,6 +4281,9 @@ def _bg_review_auto_deny(command, description, **kwargs):
                         api_key=_parent_runtime.get("api_key") or None,
                         credential_pool=getattr(self, "_credential_pool", None),
                         parent_session_id=self.session_id,
+                        session_id=self.session_id,
+                        pass_session_id=self.pass_session_id,
+                        cached_system_prompt=review_system_prompt,
                         enabled_toolsets=["memory", "skills"],
                     )
                     review_agent._memory_write_origin = "background_review"

tests/agent/test_auxiliary_client.py

@@ -2480,9 +2480,41 @@ def test_resolve_provider_client_without_explicit_api_key_falls_back_to_env(
             # Verify the env var fallback was used
             mock_openai.assert_called_once()
             call_kwargs = mock_openai.call_args[1]
-            assert call_kwargs["api_key"] == "env-fallback-key", (
-                f"Expected env fallback key to be used when explicit_api_key is None, got: {call_kwargs['api_key']}"
-            )
+        assert call_kwargs["api_key"] == "env-fallback-key", (
+            f"Expected env fallback key to be used when explicit_api_key is None, got: {call_kwargs['api_key']}"
+        )
+
+
+class TestOpenAIProxyClientConstruction:
+    """OpenAI wrapper should disable environment proxy lookup for localhost endpoints."""
+
+    def test_localhost_base_url_uses_http_client_with_trust_env_false(self, monkeypatch):
+        mock_class = MagicMock(return_value=MagicMock())
+        with patch("agent.auxiliary_client._load_openai_cls", return_value=mock_class):
+            aux_client = None
+            with patch("agent.auxiliary_client.base_url_hostname", return_value="localhost"):
+                from agent.auxiliary_client import OpenAI
+
+                aux_client = OpenAI(api_key="dummy", base_url="http://localhost:20128/v1")
+        assert aux_client is not None
+        assert mock_class.call_count == 1
+        call_kwargs = mock_class.call_args.kwargs
+        assert "http_client" in call_kwargs
+        http_client = call_kwargs["http_client"]
+        assert http_client.trust_env is False
+        http_client.close()
+        assert call_kwargs["base_url"] == "http://localhost:20128/v1"
+
+    def test_remote_base_url_does_not_inject_http_client(self, monkeypatch):
+        mock_class = MagicMock(return_value=MagicMock())
+        with patch("agent.auxiliary_client._load_openai_cls", return_value=mock_class):
+            from agent.auxiliary_client import OpenAI
+
+            OpenAI(api_key="dummy", base_url="https://api.openai.com/v1")
+        assert mock_class.call_count == 1
+        call_kwargs = mock_class.call_args.kwargs
+        assert "http_client" not in call_kwargs
+        assert call_kwargs["base_url"] == "https://api.openai.com/v1"
 
 
 class TestAnthropicExplicitApiKey:

tests/cron/test_cron_runtime_profile_paths.py

@@ -0,0 +1,38 @@
+"""Regression test for dynamic cron paths after profile override (issue #25295)."""
+
+import importlib
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent.parent))
+
+
+def test_cron_jobs_use_active_profile_after_hermes_home_switch(monkeypatch, tmp_path):
+    """Cron writes under the active profile directory after HERMES_HOME changes."""
+    hermes_root = tmp_path / ".hermes"
+    profile_dir = hermes_root / "profiles" / "coder"
+    profile_dir.mkdir(parents=True, exist_ok=True)
+    (hermes_root / "active_profile").write_text("coder")
+
+    # Import cron jobs while HERMES_HOME still points to the hermes root.
+    monkeypatch.setenv("HERMES_HOME", str(hermes_root))
+    import cron.jobs as jobs_mod
+    importlib.reload(jobs_mod)
+
+    # Simulate process startup profile resolution.
+    from hermes_cli.main import _apply_profile_override
+
+    _apply_profile_override()
+
+    # Any write should now target the profile dir, not the root path.
+    job = jobs_mod.create_job(prompt="status check", schedule="every 1h")
+    assert job["id"]
+
+    active_profile_jobs = profile_dir / "cron" / "jobs.json"
+    root_jobs = hermes_root / "cron" / "jobs.json"
+    assert active_profile_jobs.exists()
+    assert not root_jobs.exists()
+
+    # Also validate the prompt-context helper follows the same runtime path.
+    _get_output_dir = jobs_mod._get_output_dir()
+    assert _get_output_dir == profile_dir / "cron" / "output"

tests/run_agent/test_background_review.py

@@ -190,3 +190,39 @@ def close(self):
     assert captured_bg_callback[0].startswith("💾 Self-improvement review:"), (
         captured_bg_callback[0]
     )
+
+
+def test_background_review_reuses_parent_system_prompt_when_available_or_builds_it(monkeypatch):
+    captured: dict = {}
+
+    class FakeReviewAgent:
+        def __init__(self, **kwargs):
+            captured["cached_system_prompt"] = kwargs.get("cached_system_prompt")
+            self._session_messages = []
+
+        def run_conversation(self, **kwargs):
+            pass
+
+        def shutdown_memory_provider(self):
+            pass
+
+        def close(self):
+            pass
+
+    monkeypatch.setattr(run_agent_module, "AIAgent", FakeReviewAgent)
+    monkeypatch.setattr(run_agent_module.threading, "Thread", ImmediateThread)
+
+    agent = _bare_agent()
+    agent._cached_system_prompt = None
+
+    def _build_prompt():
+        return "built parent prompt"
+    agent._build_system_prompt = _build_prompt
+
+    AIAgent._spawn_background_review(
+        agent,
+        messages_snapshot=[{"role": "user", "content": "hello"}],
+        review_memory=True,
+    )
+
+    assert captured["cached_system_prompt"] == "built parent prompt"