ci: add focused test guidance signal (#4982)

* ci: add focused test guidance signal

* ci: diff focused guidance from merge base

.github/scripts/focused_test_guidance.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""Report focused pytest guidance for changed paths under tests/."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import shlex
+import subprocess
+import sys
+from collections.abc import Iterable
+from pathlib import PurePosixPath
+
+
+def parse_paths(raw_paths: bytes) -> list[str]:
+    """Decode the NUL-delimited output of ``git diff --name-only -z``."""
+    return [os.fsdecode(path) for path in raw_paths.split(b"\0") if path]
+
+
+def changed_paths_from_merge_base(base_sha: str, head_sha: str) -> list[str]:
+    """Return changed ``tests/`` paths using GitHub PR three-dot semantics.
+
+    GitHub PR changed files are based on the merge base and the PR head, not a
+    direct endpoint diff between the current base branch tip and the PR head.
+    Using the direct endpoint diff can include files changed only on the base
+    branch when the PR branch is stale.
+    """
+    merge_base = subprocess.check_output(
+        ["git", "merge-base", base_sha, head_sha],
+        stderr=subprocess.DEVNULL,
+    ).strip()
+    raw_paths = subprocess.check_output(
+        [
+            "git",
+            "diff",
+            "--name-only",
+            "--diff-filter=ACMRT",
+            "-z",
+            os.fsdecode(merge_base),
+            head_sha,
+            "--",
+            "tests/",
+        ],
+    )
+    return parse_paths(raw_paths)
+
+
+def select_test_paths(paths: Iterable[str]) -> list[str]:
+    """Return unique, repository-relative paths contained by tests/."""
+    selected: set[str] = set()
+    for raw_path in paths:
+        path = PurePosixPath(raw_path)
+        if path.is_absolute() or ".." in path.parts:
+            continue
+        parts = tuple(part for part in path.parts if part != ".")
+        if len(parts) >= 2 and parts[0] == "tests":
+            selected.add(PurePosixPath(*parts).as_posix())
+    return sorted(selected)
+
+
+def is_pytest_file(path: str) -> bool:
+    """Return whether a changed path follows this repository's pytest naming."""
+    name = PurePosixPath(path).name
+    return name.endswith(".py") and (
+        name.startswith("test_") or name.endswith("_test.py")
+    )
+
+
+def pytest_command(paths: Iterable[str]) -> str:
+    """Build a copyable pytest command for changed runnable test files."""
+    command = ["python3", "-m", "pytest", "-q", *paths]
+    return shlex.join(command)
+
+
+def format_report(paths: Iterable[str]) -> str:
+    """Format focused guidance for CI logs and the workflow summary."""
+    changed_paths = select_test_paths(paths)
+    runnable_paths = [path for path in changed_paths if is_pytest_file(path)]
+    lines = ["## Focused test guidance (report-only)", ""]
+    if not changed_paths:
+        lines.append("No changed paths under `tests/`.")
+    else:
+        lines.extend(["Changed paths under `tests/`:", ""])
+        lines.extend(f"- `{path}`" for path in changed_paths)
+    lines.extend(["", "Suggested focused validation:", ""])
+    if runnable_paths:
+        lines.append(f"```sh\n{pytest_command(runnable_paths)}\n```")
+    else:
+        lines.append("No directly runnable pytest files changed.")
+    lines.extend(
+        [
+            "",
+            "This guidance does not infer tests from source changes. "
+            "Existing blocking CI remains the source of truth.",
+        ]
+    )
+    return "\n".join(lines)
+
+
+def _parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Report focused pytest guidance for changed tests/ paths.",
+    )
+    parser.add_argument("--base-sha", help="Pull request base commit SHA.")
+    parser.add_argument("--head-sha", help="Pull request head commit SHA.")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = _parse_args(sys.argv[1:] if argv is None else argv)
+    if bool(args.base_sha) != bool(args.head_sha):
+        raise SystemExit("--base-sha and --head-sha must be provided together")
+
+    if args.base_sha and args.head_sha:
+        paths = changed_paths_from_merge_base(args.base_sha, args.head_sha)
+    else:
+        paths = parse_paths(sys.stdin.buffer.read())
+
+    print(format_report(paths))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/workflows/ci.yml

@@ -15,6 +15,60 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
+  focused-test-guidance:
+    name: Focused test guidance (report-only)
+    if: github.event_name == 'pull_request'
+    runs-on: ubuntu-latest
+    continue-on-error: true
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
+        with:
+          fetch-depth: 0
+          persist-credentials: false
+      - name: Report changed test paths
+        env:
+          BASE_SHA: ${{ github.event.pull_request.base.sha }}
+          HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+        run: |
+          report_file="$RUNNER_TEMP/focused-test-guidance.md"
+          publish_report() {
+            cat "$report_file"
+            if [ -n "${GITHUB_STEP_SUMMARY:-}" ]; then
+              cat "$report_file" >> "$GITHUB_STEP_SUMMARY" || true
+            fi
+            return 0
+          }
+
+          report_unavailable() {
+            {
+              printf '%s\n\n' '## Focused test guidance unavailable (report-only)'
+              printf '%s\n\n' "$1"
+              printf '%s\n' 'Existing blocking CI remains the source of truth.'
+            } > "$report_file"
+            publish_report
+            exit 0
+          }
+
+          if [ -z "$BASE_SHA" ] || [ -z "$HEAD_SHA" ]; then
+            report_unavailable "Pull request base/head metadata is missing."
+          fi
+
+          if ! git cat-file -e "${BASE_SHA}^{commit}" 2>/dev/null; then
+            report_unavailable "The pull request base commit is unavailable locally."
+          fi
+
+          if ! git cat-file -e "${HEAD_SHA}^{commit}" 2>/dev/null; then
+            report_unavailable "The pull request head commit is unavailable locally."
+          fi
+
+          if ! python3 .github/scripts/focused_test_guidance.py \
+            --base-sha "$BASE_SHA" \
+            --head-sha "$HEAD_SHA" > "$report_file"; then
+            report_unavailable "The focused test guidance helper could not produce a report."
+          fi
+
+          publish_report
+
   python-syntax:
     name: Python syntax (compileall)
     runs-on: ubuntu-latest

src/rag_manager.py

@@ -32,9 +32,9 @@ class RAGManager:
         logger.info("RAGManager initialized as wrapper for VectorRAG")
     
     # Delegate all methods to VectorRAG
-    def search(self, query: str, k: int = 5) -> List[Dict[str, Any]]:
+    def search(self, query: str, k: int = 5, owner: Optional[str] = None) -> List[Dict[str, Any]]:
         """Search for documents - delegates to VectorRAG."""
-        return self.vector_rag.search(query, k)
+        return self.vector_rag.search(query, k, owner=owner)
     
     def index_personal_documents(
         self,

src/tool_utils.py

@@ -54,6 +54,8 @@ def _parse_tool_args(content):
     if isinstance(content, str):
         try:
             args = json.loads(content) if content.strip() else {}
+            if not isinstance(args, dict):
+                args = {}
         except (json.JSONDecodeError, TypeError) as e:
             raise ValueError(str(e))
     elif isinstance(content, dict):

tests/run_focus.py

@@ -50,6 +50,14 @@ AREAS: tuple[str, ...] = (
 # Backward-compatible aggregate selectors for focused runs whose original
 # monolithic files were split into more specific taxonomy sub-areas.
 SUB_AREA_ALIASES: dict[str, tuple[str, ...]] = {
+    "service_health": (
+        "service_health_chromadb",
+        "service_health_search",
+        "service_health_ntfy",
+        "service_health_email",
+        "service_health_providers",
+        "service_health_collect",
+    ),
     "embedding": ("embedding", "embedding_memory"),
 }
 
@@ -214,6 +222,7 @@ def build_parser(
     """Build the argument parser for the focused runner."""
     if valid_sub_areas is None:
         valid_sub_areas = discover_sub_areas()
+    valid_sub_areas = frozenset(valid_sub_areas) | frozenset(SUB_AREA_ALIASES)
     parser = argparse.ArgumentParser(
         prog="run_focus.py",
         description=(

tests/test_admin_tools_registry.py

@@ -72,3 +72,8 @@ def test_parse_tool_args_lives_in_tool_utils_single_source():
     assert _parse_tool_args('{"action":"add"}') == {"action": "add"}
     # body-envelope unwrap still works
     assert _parse_tool_args('{"body":{"action":"x"}}') == {"action": "x"}
+
+    # non-dict JSON values should return {}
+    assert _parse_tool_args('[1, 2]') == {}
+    assert _parse_tool_args('42') == {}
+    assert _parse_tool_args('"hello"') == {}

tests/test_focused_test_guidance.py

@@ -0,0 +1,148 @@
+"""Tests for the report-only changed-test guidance helper."""
+
+from __future__ import annotations
+
+import importlib.util
+import subprocess
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+SCRIPT_PATH = ROOT / ".github" / "scripts" / "focused_test_guidance.py"
+
+
+def _load_helper():
+    spec = importlib.util.spec_from_file_location("focused_test_guidance", SCRIPT_PATH)
+    assert spec is not None
+    assert spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+guidance = _load_helper()
+
+
+def test_parse_paths_supports_nul_delimited_git_output():
+    raw_paths = b"tests/test_alpha.py\0tests/path with spaces/test_beta.py\0"
+
+    assert guidance.parse_paths(raw_paths) == [
+        "tests/test_alpha.py",
+        "tests/path with spaces/test_beta.py",
+    ]
+
+
+def test_select_test_paths_ignores_paths_outside_tests():
+    paths = [
+        "src/test_alpha.py",
+        "tests/test_beta.py",
+        "./tests/unit/example_test.py",
+        "tests/../src/test_gamma.py",
+    ]
+
+    assert guidance.select_test_paths(paths) == [
+        "tests/test_beta.py",
+        "tests/unit/example_test.py",
+    ]
+
+
+def test_select_test_paths_deduplicates_paths():
+    paths = ["tests/test_alpha.py", "./tests/test_alpha.py"]
+
+    assert guidance.select_test_paths(paths) == ["tests/test_alpha.py"]
+
+
+def test_format_report_builds_command_for_changed_python_test():
+    report = guidance.format_report(["tests/test_beta.py"])
+
+    assert "- `tests/test_beta.py`" in report
+    assert "python3 -m pytest -q tests/test_beta.py" in report
+    assert "does not infer tests from source changes" in report
+    assert "Existing blocking CI remains the source of truth" in report
+
+
+def test_format_report_lists_changed_non_python_test_path_without_command():
+    report = guidance.format_report(["tests/README.md"])
+
+    assert "- `tests/README.md`" in report
+    assert "No directly runnable pytest files changed." in report
+    assert "python3 -m pytest" not in report
+
+
+def test_format_report_ignores_src_path():
+    report = guidance.format_report(["src/test_ignored.py"])
+
+    assert "src/test_ignored.py" not in report
+    assert "No changed paths under `tests/`." in report
+
+
+def test_format_report_shell_quotes_path_with_spaces():
+    report = guidance.format_report(["tests/path with spaces/test_beta.py"])
+
+    assert "- `tests/path with spaces/test_beta.py`" in report
+    assert (
+        "python3 -m pytest -q 'tests/path with spaces/test_beta.py'"
+    ) in report
+
+
+def test_format_report_handles_no_changed_test_paths():
+    report = guidance.format_report([])
+
+    assert "No changed paths under `tests/`." in report
+    assert "No directly runnable pytest files changed." in report
+
+def _git(repo: Path, *args: str) -> str:
+    return subprocess.check_output(["git", *args], cwd=repo, text=True).strip()
+
+
+def _write(path: Path, content: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(content, encoding="utf-8")
+
+
+def test_changed_paths_from_merge_base_excludes_base_only_test_changes(tmp_path, monkeypatch):
+    repo = tmp_path / "repo"
+    repo.mkdir()
+
+    _git(repo, "init")
+    _git(repo, "config", "user.email", "ci@example.test")
+    _git(repo, "config", "user.name", "CI Test")
+
+    _write(repo / "tests/test_shared.py", "def test_shared():\n    assert True\n")
+    _git(repo, "add", "tests/test_shared.py")
+    _git(repo, "commit", "-m", "base")
+    ancestor = _git(repo, "rev-parse", "HEAD")
+
+    _git(repo, "checkout", "-b", "feature")
+    _write(repo / "tests/test_pr_delta.py", "def test_pr_delta():\n    assert True\n")
+    _git(repo, "add", "tests/test_pr_delta.py")
+    _git(repo, "commit", "-m", "add pr test")
+    head_sha = _git(repo, "rev-parse", "HEAD")
+
+    _git(repo, "checkout", "-b", "dev", ancestor)
+    _write(repo / "tests/test_shared.py", "def test_shared():\n    assert 1 == 1\n")
+    _git(repo, "add", "tests/test_shared.py")
+    _git(repo, "commit", "-m", "base-only test change")
+    base_sha = _git(repo, "rev-parse", "HEAD")
+
+    endpoint_paths = guidance.parse_paths(
+        subprocess.check_output(
+            [
+                "git",
+                "diff",
+                "--name-only",
+                "--diff-filter=ACMRT",
+                "-z",
+                base_sha,
+                head_sha,
+                "--",
+                "tests/",
+            ],
+            cwd=repo,
+        )
+    )
+    assert "tests/test_shared.py" in endpoint_paths
+
+    monkeypatch.chdir(repo)
+    assert guidance.changed_paths_from_merge_base(base_sha, head_sha) == [
+        "tests/test_pr_delta.py"
+    ]

tests/test_rag_search_signature.py

@@ -0,0 +1,22 @@
+import unittest
+from unittest.mock import MagicMock, patch
+from src.rag_manager import RAGManager
+
+class TestRAGManagerSearchSignature(unittest.TestCase):
+    @patch('src.rag_manager.VectorRAG')
+    def test_search_signature_accepts_owner(self, mock_vector_rag_class):
+        # Create a mock instance for VectorRAG
+        mock_vector_rag = MagicMock()
+        mock_vector_rag_class.return_value = mock_vector_rag
+
+        # Initialize RAGManager
+        manager = RAGManager()
+
+        # Test call with owner parameter
+        manager.search("test query", k=3, owner="user1")
+
+        # Verify that search was called on the underlying vector_rag with the correct parameters
+        mock_vector_rag.search.assert_called_once_with("test query", 3, owner="user1")
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_run_focus.py

@@ -448,3 +448,45 @@ def test_fast_lane_collects_only_unmarked_auth_concurrency_test():
     assert _FAST_AUTH_CONCURRENCY_TEST in collected
     for slow_test in _SLOW_AUTH_CONCURRENCY_TESTS:
         assert slow_test not in collected, f"slow test was not deselected: {slow_test}"
+
+def test_service_health_sub_area_command_includes_split_files():
+    assert _cmd(sub_area="service_health") == [
+        PY,
+        "-m",
+        "pytest",
+        "-m",
+        (
+            "(sub_service_health_chromadb or "
+            "sub_service_health_search or "
+            "sub_service_health_ntfy or "
+            "sub_service_health_email or "
+            "sub_service_health_providers or "
+            "sub_service_health_collect)"
+        ),
+    ]
+
+
+def test_service_health_alias_is_accepted_by_run():
+    seen = []
+
+    def executor(cmd):
+        seen.append(cmd)
+        return 0
+
+    result = run(["--sub-area", "service_health"], executor=executor)
+
+    assert result == 0
+    assert len(seen) == 1
+    assert seen[0][1:] == [
+        "-m",
+        "pytest",
+        "-m",
+        (
+            "(sub_service_health_chromadb or "
+            "sub_service_health_search or "
+            "sub_service_health_ntfy or "
+            "sub_service_health_email or "
+            "sub_service_health_providers or "
+            "sub_service_health_collect)"
+        ),
+    ]

tests/test_service_health.py

@@ -1,472 +0,0 @@
-"""Tests for src.service_health — the consolidated degraded-state report.
-
-Imports the real module (conftest.py stubs the heavy deps). Network is never
-touched: HTTP probes take an injected `http_get`, and the email/provider probes
-take an injected `connect` / `probe`. Asserts the ok/degraded/down/disabled
-mapping per subsystem, the overall rollup, and that no secrets leak into meta.
-"""
-import types
-
-import pytest
-
-from src import service_health as sh
-
-
-def _resp(status_code):
-    return types.SimpleNamespace(status_code=status_code)
-
-
-def _raise(*_a, **_k):
-    raise RuntimeError("connection refused")
-
-
-# ── chromadb_health ──
-
-class _Store:
-    def __init__(self, healthy):
-        self.healthy = healthy
-
-
-def test_chromadb_both_healthy_ok():
-    s = sh.chromadb_health(_Store(True), _Store(True))
-    assert s["status"] == sh.OK
-    assert s["meta"] == {"rag": True, "memory": True}
-
-
-def test_chromadb_one_down_degraded():
-    s = sh.chromadb_health(_Store(True), _Store(False))
-    assert s["status"] == sh.DEGRADED
-
-
-def test_chromadb_both_unhealthy_down():
-    s = sh.chromadb_health(_Store(False), _Store(False))
-    assert s["status"] == sh.DOWN
-
-
-def test_chromadb_both_absent_disabled():
-    s = sh.chromadb_health(None, None)
-    assert s["status"] == sh.DISABLED
-
-
-def test_chromadb_one_absent_one_healthy_ok():
-    # An absent store is not a failure; the present one being healthy is ok.
-    s = sh.chromadb_health(_Store(True), None)
-    assert s["status"] == sh.OK
-    assert s["meta"]["memory"] is None
-
-
-# ── searxng_health ──
-
-def test_searxng_disabled_when_other_provider():
-    s = sh.searxng_health({"search_provider": "brave"})
-    assert s["status"] == sh.DISABLED
-
-
-def test_searxng_ok_on_healthz():
-    s = sh.searxng_health(
-        {"search_provider": "searxng", "search_url": "http://sx:8080"},
-        http_get=lambda url, timeout: _resp(200),
-    )
-    assert s["status"] == sh.OK
-    assert s["meta"]["probed"] == "/healthz"
-
-
-def test_searxng_ok_on_root_fallback():
-    def getter(url, timeout):
-        return _resp(404) if url.endswith("/healthz") else _resp(200)
-
-    s = sh.searxng_health(
-        {"search_provider": "searxng", "search_url": "http://sx:8080"},
-        http_get=getter,
-    )
-    assert s["status"] == sh.OK
-    assert s["meta"]["probed"] == "/"
-
-
-def test_searxng_down_on_exception():
-    s = sh.searxng_health(
-        {"search_provider": "searxng", "search_url": "http://sx:8080"},
-        http_get=_raise,
-    )
-    assert s["status"] == sh.DOWN
-
-
-def test_searxng_down_on_5xx():
-    s = sh.searxng_health(
-        {"search_provider": "searxng", "search_url": "http://sx:8080"},
-        http_get=lambda url, timeout: _resp(502),
-    )
-    assert s["status"] == sh.DOWN
-
-
-# ── ntfy_health ──
-
-def _ntfy_intg():
-    return [{"preset": "ntfy", "enabled": True, "base_url": "http://ntfy:80"}]
-
-
-def test_ntfy_disabled_without_integration():
-    s = sh.ntfy_health([], {"reminder_channel": "ntfy"})
-    assert s["status"] == sh.DISABLED
-
-
-def test_ntfy_ok():
-    s = sh.ntfy_health(_ntfy_intg(), {"reminder_channel": "ntfy"},
-                       http_get=lambda url, timeout: _resp(200))
-    assert s["status"] == sh.OK
-    assert s["meta"]["base"] == "http://ntfy:80"
-
-
-def test_ntfy_probes_v1_health_not_a_topic():
-    seen = {}
-
-    def getter(url, timeout):
-        seen["url"] = url
-        return _resp(200)
-
-    sh.ntfy_health(_ntfy_intg(), {"reminder_channel": "ntfy"}, http_get=getter)
-    # Non-intrusive: hits /v1/health, never publishes to a topic.
-    assert seen["url"].endswith("/v1/health")
-
-
-def test_ntfy_down_on_exception():
-    s = sh.ntfy_health(_ntfy_intg(), {"reminder_channel": "ntfy"},
-                       http_get=_raise)
-    assert s["status"] == sh.DOWN
-
-
-# ── email_health ──
-
-def _acct(name, host="imap.example.com"):
-    return {"account_id": name, "account_name": name, "imap_host": host,
-            "imap_password": "hunter2"}
-
-
-class _Conn:
-    def logout(self):
-        pass
-
-
-def test_email_disabled_without_accounts():
-    assert sh.email_health([])["status"] == sh.DISABLED
-
-
-def test_email_ok_all_connect():
-    s = sh.email_health([_acct("a"), _acct("b")], connect=lambda _id: _Conn())
-    assert s["status"] == sh.OK
-
-
-def test_email_degraded_some_fail():
-    def connect(account_id):
-        if account_id == "bad":
-            raise RuntimeError("auth failed")
-        return _Conn()
-
-    s = sh.email_health([_acct("good"), _acct("bad")], connect=connect)
-    assert s["status"] == sh.DEGRADED
-
-
-def test_email_down_all_fail():
-    s = sh.email_health([_acct("a")], connect=_raise)
-    assert s["status"] == sh.DOWN
-
-
-def test_email_account_without_host_marked_failed():
-    s = sh.email_health([_acct("a", host="")], connect=lambda _id: _Conn())
-    assert s["status"] == sh.DOWN
-
-
-def test_email_meta_never_leaks_password():
-    s = sh.email_health([_acct("a")], connect=lambda _id: _Conn())
-    assert "hunter2" not in repr(s)
-
-
-# ── providers_health ──
-
-def _ep(name):
-    return {"name": name, "base_url": f"http://{name}:8000/v1", "api_key": "sk-secret"}
-
-
-def test_providers_disabled_without_endpoints():
-    assert sh.providers_health([])["status"] == sh.DISABLED
-
-
-def test_providers_ok_all_reachable():
-    s = sh.providers_health([_ep("a")],
-                            probe=lambda base, key, timeout: ["m1", "m2"])
-    assert s["status"] == sh.OK
-    assert s["meta"]["endpoints"][0]["model_count"] == 2
-
-
-def test_providers_degraded_some_empty():
-    def probe(base, key, timeout):
-        return ["m1"] if "good" in base else []
-
-    s = sh.providers_health([_ep("good"), _ep("bad")], probe=probe)
-    assert s["status"] == sh.DEGRADED
-
-
-def test_providers_down_all_fail():
-    s = sh.providers_health([_ep("a")], probe=_raise)
-    assert s["status"] == sh.DOWN
-
-
-def test_providers_meta_never_leaks_api_key():
-    s = sh.providers_health([_ep("a")],
-                            probe=lambda base, key, timeout: ["m1"])
-    assert "sk-secret" not in repr(s)
-
-
-# ── rollup ──
-
-def test_rollup_picks_worst_non_disabled():
-    services = [
-        {"status": sh.OK}, {"status": sh.DISABLED},
-        {"status": sh.DEGRADED}, {"status": sh.OK},
-    ]
-    assert sh._rollup(services) == sh.DEGRADED
-
-
-def test_rollup_down_beats_degraded():
-    assert sh._rollup([{"status": sh.DEGRADED}, {"status": sh.DOWN}]) == sh.DOWN
-
-
-def test_rollup_all_disabled_is_ok():
-    assert sh._rollup([{"status": sh.DISABLED}, {"status": sh.DISABLED}]) == sh.OK
-
-
-# ── collect_service_health (async aggregate) ──
-
-def test_collect_service_health_shape(monkeypatch):
-    import asyncio
-
-    # Avoid touching real data sources / network.
-    monkeypatch.setattr(sh, "_gather_inputs", lambda: {
-        "settings": {"search_provider": "disabled"},
-        "integrations": [],
-        "accounts": [],
-        "endpoints": [],
-    })
-    out = asyncio.run(sh.collect_service_health(_Store(True), _Store(True)))
-    assert set(out) == {"overall", "services", "timestamp"}
-    names = {s["name"] for s in out["services"]}
-    assert names == {"chromadb", "searxng", "ntfy", "email", "providers"}
-    # Chroma healthy, everything else disabled → overall ok.
-    assert out["overall"] == sh.OK
-
-
-# ── _safe_url: strip userinfo / query / fragment ──
-
-@pytest.mark.parametrize("raw,expected", [
-    ("http://user:pass@host:8080/path?api_key=secret#frag", "http://host:8080/path"),
-    ("https://admin:hunter2@searx.example.com/", "https://searx.example.com"),
-    ("http://ntfy.local:80?token=abc", "http://ntfy.local:80"),
-    ("host:8080", "host:8080"),
-    ("", ""),
-    (None, ""),
-])
-def test_safe_url_strips_secrets(raw, expected):
-    out = sh._safe_url(raw)
-    assert out == expected
-    for bad in ("pass", "secret", "hunter2", "abc", "token", "@"):
-        if raw and bad in raw and bad not in expected:
-            assert bad not in out
-
-
-# ── _classify_error: controlled categories, never raw text ──
-
-def test_classify_error_categories():
-    import socket
-    assert sh._classify_error(TimeoutError()) == "timeout"
-    assert sh._classify_error(socket.timeout()) == "timeout"
-    assert sh._classify_error(socket.gaierror()) == "dns_error"
-    assert sh._classify_error(ConnectionRefusedError()) == "connection_refused"
-    assert sh._classify_error(OSError("boom")) == "network_error"
-    assert sh._classify_error(ValueError("x")) == "error"
-
-
-# ── Sanitization in subsystem output (blocker #2) ──
-
-def test_searxng_meta_redacts_instance_url():
-    s = sh.searxng_health(
-        {"search_provider": "searxng",
-         "search_url": "http://user:s3cr3t@searx.local:8080/?token=zzz"},
-        http_get=lambda url, timeout: _resp(200),
-    )
-    blob = repr(s)
-    assert "s3cr3t" not in blob and "zzz" not in blob and "user:" not in blob
-    assert s["meta"]["instance"] == "http://searx.local:8080"
-
-
-def test_searxng_down_uses_error_category_not_raw_exception():
-    def boom(url, timeout):
-        raise RuntimeError("failed connecting to http://user:pw@searx.local secret-token")
-    s = sh.searxng_health(
-        {"search_provider": "searxng", "search_url": "http://searx.local"},
-        http_get=boom,
-    )
-    assert s["status"] == sh.DOWN
-    assert s["meta"]["error"] == "error"           # controlled category token
-    assert "secret-token" not in repr(s) and "pw@" not in repr(s)
-
-
-def test_ntfy_meta_redacts_userinfo_in_base():
-    intg = [{"preset": "ntfy", "enabled": True,
-             "base_url": "https://user:topsecret@ntfy.example.com"}]
-    seen = {}
-
-    def getter(url, timeout):
-        seen["url"] = url          # the probe itself may keep credentials
-        return _resp(200)
-
-    s = sh.ntfy_health(intg, {"reminder_channel": "ntfy"}, http_get=getter)
-    assert s["meta"]["base"] == "https://ntfy.example.com"
-    assert "topsecret" not in repr(s)
-
-
-def test_providers_name_fallback_is_sanitized():
-    # No display name → falls back to the base_url, which must be sanitized.
-    ep = {"base_url": "http://user:k3y@prov.local:9000/v1?api_key=zzz", "api_key": "sk-x"}
-    s = sh.providers_health([ep], probe=lambda b, k, t: ["m1"])
-    entry = s["meta"]["endpoints"][0]
-    assert entry["name"] == "http://prov.local:9000/v1"
-    assert "k3y" not in repr(s) and "zzz" not in repr(s) and "sk-x" not in repr(s)
-
-
-def test_providers_probe_exception_maps_to_category():
-    def boom(base, key, timeout):
-        raise RuntimeError(f"500 from {base} with key {key}")  # would leak base+key
-    s = sh.providers_health([_ep("a")], probe=boom)
-    assert s["status"] == sh.DOWN
-    assert s["meta"]["endpoints"][0]["error"] == "error"
-    assert "sk-secret" not in repr(s) and "http://a" not in repr(s)
-
-
-def test_email_connect_exception_maps_to_category():
-    def boom(account_id):
-        raise RuntimeError("login failed for user bob with password hunter2")
-    s = sh.email_health([_acct("a")], connect=boom)
-    assert s["status"] == sh.DOWN
-    assert s["meta"]["accounts"][0]["error"] == "error"
-    assert "hunter2" not in repr(s)
-
-
-# ── Bounded wall-clock (blocker #1) ──
-
-def test_providers_bounded_marks_slow_as_timeout(monkeypatch):
-    import time
-    monkeypatch.setattr(sh, "_FANOUT_BUDGET", 1)
-
-    def probe(base, key, timeout):
-        if "slow" in base:
-            time.sleep(10)          # would blow the budget if unbounded
-        return ["m1"]
-
-    eps = [{"name": "fast", "base_url": "http://fast", "api_key": "k"},
-           {"name": "slow", "base_url": "http://slow", "api_key": "k"}]
-    t0 = time.monotonic()
-    out = sh.providers_health(eps, probe=probe)
-    elapsed = time.monotonic() - t0
-    assert elapsed < 4, f"providers_health not bounded: took {elapsed:.1f}s"
-    by = {e["name"]: e for e in out["meta"]["endpoints"]}
-    assert by["fast"]["ok"] is True
-    assert by["slow"]["ok"] is False and by["slow"]["error"] == "timeout"
-    assert out["status"] == sh.DEGRADED
-
-
-def test_providers_bounded_with_many_slow_endpoints(monkeypatch):
-    import time
-    monkeypatch.setattr(sh, "_FANOUT_BUDGET", 1)
-
-    def probe(base, key, timeout):
-        time.sleep(10)
-        return ["m1"]
-
-    eps = [{"name": f"ep{i}", "base_url": f"http://ep{i}", "api_key": "k"}
-           for i in range(25)]
-    t0 = time.monotonic()
-    out = sh.providers_health(eps, probe=probe)
-    elapsed = time.monotonic() - t0
-    # 25 endpoints * sleep would be huge if sequential; bounded keeps it ~budget.
-    assert elapsed < 4, f"not bounded with many endpoints: {elapsed:.1f}s"
-    assert out["status"] == sh.DOWN
-    assert all(e["error"] == "timeout" for e in out["meta"]["endpoints"])
-
-
-def test_email_bounded_marks_slow_as_timeout(monkeypatch):
-    import time
-    monkeypatch.setattr(sh, "_FANOUT_BUDGET", 1)
-
-    def connect(account_id):
-        if account_id == "slow":
-            time.sleep(10)
-        return _Conn()
-
-    accts = [_acct("fast"), _acct("slow")]
-    accts[1]["account_id"] = "slow"
-    t0 = time.monotonic()
-    out = sh.email_health(accts, connect=connect)
-    elapsed = time.monotonic() - t0
-    assert elapsed < 4, f"email_health not bounded: took {elapsed:.1f}s"
-    by = {a["name"]: a for a in out["meta"]["accounts"]}
-    assert by["slow"]["error"] == "timeout"
-
-
-def test_collect_runs_subsystems_concurrently(monkeypatch):
-    # The aggregate is bounded by running the (internally-bounded) subsystems
-    # concurrently, so total wall-clock ≈ max(subsystem), not the sum. Each of
-    # the four network subsystems here sleeps ~0.6s; sequential would be ~2.4s.
-    import asyncio
-    import time
-    monkeypatch.setattr(sh, "_gather_inputs", lambda: {
-        "settings": {}, "integrations": [], "accounts": [], "endpoints": [],
-    })
-
-    def slow(name):
-        def _fn(*_a, **_k):
-            time.sleep(0.6)
-            return {"name": name, "status": sh.OK, "detail": "", "meta": {}}
-        return _fn
-
-    monkeypatch.setattr(sh, "searxng_health", slow("searxng"))
-    monkeypatch.setattr(sh, "ntfy_health", slow("ntfy"))
-    monkeypatch.setattr(sh, "email_health", slow("email"))
-    monkeypatch.setattr(sh, "providers_health", slow("providers"))
-
-    t0 = time.monotonic()
-    out = asyncio.run(sh.collect_service_health(None, None))
-    elapsed = time.monotonic() - t0
-    assert elapsed < 1.5, f"subsystems not concurrent: took {elapsed:.1f}s"
-    assert {s["name"] for s in out["services"]} == {
-        "chromadb", "searxng", "ntfy", "email", "providers"}
-
-
-def test_collect_aggregate_deadline_yields_controlled_result(monkeypatch):
-    # If the gather overruns the aggregate ceiling, the response is still a
-    # controlled {overall, services, timestamp} with each network subsystem
-    # marked down/timeout — never a hang or a raised exception.
-    import asyncio
-    import time
-    monkeypatch.setattr(sh, "_AGGREGATE_DEADLINE", 0.5)
-    monkeypatch.setattr(sh, "_SUBSYSTEM_DEADLINE", 0.4)
-    monkeypatch.setattr(sh, "_gather_inputs", lambda: {
-        "settings": {}, "integrations": [], "accounts": [], "endpoints": [],
-    })
-
-    async def _slow_gather(*coros, **_k):
-        for c in coros:                 # close unawaited coros to avoid warnings
-            close = getattr(c, "close", None)
-            if close:
-                close()
-        await asyncio.sleep(5)
-
-    # Force the outer wait_for to trip by making gather itself slow.
-    monkeypatch.setattr(sh.asyncio, "gather", _slow_gather)
-    t0 = time.monotonic()
-    out = asyncio.run(sh.collect_service_health(None, None))
-    elapsed = time.monotonic() - t0
-    assert elapsed < 2, f"aggregate deadline did not bound: {elapsed:.1f}s"
-    assert set(out) == {"overall", "services", "timestamp"}
-    net = [s for s in out["services"] if s["name"] != "chromadb"]
-    assert all(s["status"] == sh.DOWN and s["meta"].get("error") == "timeout"
-               for s in net)

tests/test_service_health_chromadb.py

@@ -0,0 +1,37 @@
+"""Tests for chromadb_health — ok/degraded/down/disabled classification."""
+import pytest
+
+from src import service_health as sh
+
+
+class _Store:
+    def __init__(self, healthy):
+        self.healthy = healthy
+
+
+def test_chromadb_both_healthy_ok():
+    s = sh.chromadb_health(_Store(True), _Store(True))
+    assert s["status"] == sh.OK
+    assert s["meta"] == {"rag": True, "memory": True}
+
+
+def test_chromadb_one_down_degraded():
+    s = sh.chromadb_health(_Store(True), _Store(False))
+    assert s["status"] == sh.DEGRADED
+
+
+def test_chromadb_both_unhealthy_down():
+    s = sh.chromadb_health(_Store(False), _Store(False))
+    assert s["status"] == sh.DOWN
+
+
+def test_chromadb_both_absent_disabled():
+    s = sh.chromadb_health(None, None)
+    assert s["status"] == sh.DISABLED
+
+
+def test_chromadb_one_absent_one_healthy_ok():
+    # An absent store is not a failure; the present one being healthy is ok.
+    s = sh.chromadb_health(_Store(True), None)
+    assert s["status"] == sh.OK
+    assert s["meta"]["memory"] is None

tests/test_service_health_collect.py

@@ -0,0 +1,139 @@
+"""Tests for rollup logic, aggregate collection, and shared utility helpers (_safe_url, _classify_error)."""
+import pytest
+
+from src import service_health as sh
+
+
+class _Store:
+    def __init__(self, healthy):
+        self.healthy = healthy
+
+
+# ── rollup ──
+
+def test_rollup_picks_worst_non_disabled():
+    services = [
+        {"status": sh.OK}, {"status": sh.DISABLED},
+        {"status": sh.DEGRADED}, {"status": sh.OK},
+    ]
+    assert sh._rollup(services) == sh.DEGRADED
+
+
+def test_rollup_down_beats_degraded():
+    assert sh._rollup([{"status": sh.DEGRADED}, {"status": sh.DOWN}]) == sh.DOWN
+
+
+def test_rollup_all_disabled_is_ok():
+    assert sh._rollup([{"status": sh.DISABLED}, {"status": sh.DISABLED}]) == sh.OK
+
+
+# ── collect_service_health (async aggregate) ──
+
+def test_collect_service_health_shape(monkeypatch):
+    import asyncio
+
+    # Avoid touching real data sources / network.
+    monkeypatch.setattr(sh, "_gather_inputs", lambda: {
+        "settings": {"search_provider": "disabled"},
+        "integrations": [],
+        "accounts": [],
+        "endpoints": [],
+    })
+    out = asyncio.run(sh.collect_service_health(_Store(True), _Store(True)))
+    assert set(out) == {"overall", "services", "timestamp"}
+    names = {s["name"] for s in out["services"]}
+    assert names == {"chromadb", "searxng", "ntfy", "email", "providers"}
+    # Chroma healthy, everything else disabled → overall ok.
+    assert out["overall"] == sh.OK
+
+
+# ── _safe_url: strip userinfo / query / fragment ──
+
+@pytest.mark.parametrize("raw,expected", [
+    ("http://user:pass@host:8080/path?api_key=secret#frag", "http://host:8080/path"),
+    ("https://admin:hunter2@searx.example.com/", "https://searx.example.com"),
+    ("http://ntfy.local:80?token=abc", "http://ntfy.local:80"),
+    ("host:8080", "host:8080"),
+    ("", ""),
+    (None, ""),
+])
+def test_safe_url_strips_secrets(raw, expected):
+    out = sh._safe_url(raw)
+    assert out == expected
+    for bad in ("pass", "secret", "hunter2", "abc", "token", "@"):
+        if raw and bad in raw and bad not in expected:
+            assert bad not in out
+
+
+# ── _classify_error: controlled categories, never raw text ──
+
+def test_classify_error_categories():
+    import socket
+    assert sh._classify_error(TimeoutError()) == "timeout"
+    assert sh._classify_error(socket.timeout()) == "timeout"
+    assert sh._classify_error(socket.gaierror()) == "dns_error"
+    assert sh._classify_error(ConnectionRefusedError()) == "connection_refused"
+    assert sh._classify_error(OSError("boom")) == "network_error"
+    assert sh._classify_error(ValueError("x")) == "error"
+
+
+# ── Concurrent collection and aggregate deadline ──
+
+def test_collect_runs_subsystems_concurrently(monkeypatch):
+    # The aggregate is bounded by running the (internally-bounded) subsystems
+    # concurrently, so total wall-clock ≈ max(subsystem), not the sum. Each of
+    # the four network subsystems here sleeps ~0.6s; sequential would be ~2.4s.
+    import asyncio
+    import time
+    monkeypatch.setattr(sh, "_gather_inputs", lambda: {
+        "settings": {}, "integrations": [], "accounts": [], "endpoints": [],
+    })
+
+    def slow(name):
+        def _fn(*_a, **_k):
+            time.sleep(0.6)
+            return {"name": name, "status": sh.OK, "detail": "", "meta": {}}
+        return _fn
+
+    monkeypatch.setattr(sh, "searxng_health", slow("searxng"))
+    monkeypatch.setattr(sh, "ntfy_health", slow("ntfy"))
+    monkeypatch.setattr(sh, "email_health", slow("email"))
+    monkeypatch.setattr(sh, "providers_health", slow("providers"))
+
+    t0 = time.monotonic()
+    out = asyncio.run(sh.collect_service_health(None, None))
+    elapsed = time.monotonic() - t0
+    assert elapsed < 1.5, f"subsystems not concurrent: took {elapsed:.1f}s"
+    assert {s["name"] for s in out["services"]} == {
+        "chromadb", "searxng", "ntfy", "email", "providers"}
+
+
+def test_collect_aggregate_deadline_yields_controlled_result(monkeypatch):
+    # If the gather overruns the aggregate ceiling, the response is still a
+    # controlled {overall, services, timestamp} with each network subsystem
+    # marked down/timeout — never a hang or a raised exception.
+    import asyncio
+    import time
+    monkeypatch.setattr(sh, "_AGGREGATE_DEADLINE", 0.5)
+    monkeypatch.setattr(sh, "_SUBSYSTEM_DEADLINE", 0.4)
+    monkeypatch.setattr(sh, "_gather_inputs", lambda: {
+        "settings": {}, "integrations": [], "accounts": [], "endpoints": [],
+    })
+
+    async def _slow_gather(*coros, **_k):
+        for c in coros:                 # close unawaited coros to avoid warnings
+            close = getattr(c, "close", None)
+            if close:
+                close()
+        await asyncio.sleep(5)
+
+    # Force the outer wait_for to trip by making gather itself slow.
+    monkeypatch.setattr(sh.asyncio, "gather", _slow_gather)
+    t0 = time.monotonic()
+    out = asyncio.run(sh.collect_service_health(None, None))
+    elapsed = time.monotonic() - t0
+    assert elapsed < 2, f"aggregate deadline did not bound: {elapsed:.1f}s"
+    assert set(out) == {"overall", "services", "timestamp"}
+    net = [s for s in out["services"] if s["name"] != "chromadb"]
+    assert all(s["status"] == sh.DOWN and s["meta"].get("error") == "timeout"
+               for s in net)

tests/test_service_health_email.py

@@ -0,0 +1,80 @@
+"""Tests for email_health — probe logic, status classification, sanitization, and bounded timeout."""
+import pytest
+
+from src import service_health as sh
+
+
+def _raise(*_a, **_k):
+    raise RuntimeError("connection refused")
+
+
+def _acct(name, host="imap.example.com"):
+    return {"account_id": name, "account_name": name, "imap_host": host,
+            "imap_password": "hunter2"}
+
+
+class _Conn:
+    def logout(self):
+        pass
+
+
+def test_email_disabled_without_accounts():
+    assert sh.email_health([])["status"] == sh.DISABLED
+
+
+def test_email_ok_all_connect():
+    s = sh.email_health([_acct("a"), _acct("b")], connect=lambda _id: _Conn())
+    assert s["status"] == sh.OK
+
+
+def test_email_degraded_some_fail():
+    def connect(account_id):
+        if account_id == "bad":
+            raise RuntimeError("auth failed")
+        return _Conn()
+
+    s = sh.email_health([_acct("good"), _acct("bad")], connect=connect)
+    assert s["status"] == sh.DEGRADED
+
+
+def test_email_down_all_fail():
+    s = sh.email_health([_acct("a")], connect=_raise)
+    assert s["status"] == sh.DOWN
+
+
+def test_email_account_without_host_marked_failed():
+    s = sh.email_health([_acct("a", host="")], connect=lambda _id: _Conn())
+    assert s["status"] == sh.DOWN
+
+
+def test_email_meta_never_leaks_password():
+    s = sh.email_health([_acct("a")], connect=lambda _id: _Conn())
+    assert "hunter2" not in repr(s)
+
+
+def test_email_connect_exception_maps_to_category():
+    def boom(account_id):
+        raise RuntimeError("login failed for user bob with password hunter2")
+    s = sh.email_health([_acct("a")], connect=boom)
+    assert s["status"] == sh.DOWN
+    assert s["meta"]["accounts"][0]["error"] == "error"
+    assert "hunter2" not in repr(s)
+
+
+def test_email_bounded_marks_slow_as_timeout(monkeypatch):
+    import time
+    monkeypatch.setattr(sh, "_FANOUT_BUDGET", 1)
+
+    def connect(account_id):
+        if account_id == "slow":
+            time.sleep(10)
+        return _Conn()
+
+    accts = [_acct("fast"), _acct("slow")]
+    accts[1]["account_id"] = "slow"
+    t0 = time.monotonic()
+    out = sh.email_health(accts, connect=connect)
+    elapsed = time.monotonic() - t0
+    assert elapsed < 4, f"email_health not bounded: took {elapsed:.1f}s"
+    by = {a["name"]: a for a in out["meta"]["accounts"]}
+    assert by["slow"]["error"] == "timeout"

tests/test_service_health_ntfy.py

@@ -0,0 +1,62 @@
+"""Tests for ntfy_health — probe logic, status classification, and sanitization."""
+import types
+
+import pytest
+
+from src import service_health as sh
+
+
+def _resp(status_code):
+    return types.SimpleNamespace(status_code=status_code)
+
+
+def _raise(*_a, **_k):
+    raise RuntimeError("connection refused")
+
+
+def _ntfy_intg():
+    return [{"preset": "ntfy", "enabled": True, "base_url": "http://ntfy:80"}]
+
+
+def test_ntfy_disabled_without_integration():
+    s = sh.ntfy_health([], {"reminder_channel": "ntfy"})
+    assert s["status"] == sh.DISABLED
+
+
+def test_ntfy_ok():
+    s = sh.ntfy_health(_ntfy_intg(), {"reminder_channel": "ntfy"},
+                       http_get=lambda url, timeout: _resp(200))
+    assert s["status"] == sh.OK
+    assert s["meta"]["base"] == "http://ntfy:80"
+
+
+def test_ntfy_probes_v1_health_not_a_topic():
+    seen = {}
+
+    def getter(url, timeout):
+        seen["url"] = url
+        return _resp(200)
+
+    sh.ntfy_health(_ntfy_intg(), {"reminder_channel": "ntfy"}, http_get=getter)
+    # Non-intrusive: hits /v1/health, never publishes to a topic.
+    assert seen["url"].endswith("/v1/health")
+
+
+def test_ntfy_down_on_exception():
+    s = sh.ntfy_health(_ntfy_intg(), {"reminder_channel": "ntfy"},
+                       http_get=_raise)
+    assert s["status"] == sh.DOWN
+
+
+def test_ntfy_meta_redacts_userinfo_in_base():
+    intg = [{"preset": "ntfy", "enabled": True,
+             "base_url": "https://user:topsecret@ntfy.example.com"}]
+    seen = {}
+
+    def getter(url, timeout):
+        seen["url"] = url          # the probe itself may keep credentials
+        return _resp(200)
+
+    s = sh.ntfy_health(intg, {"reminder_channel": "ntfy"}, http_get=getter)
+    assert s["meta"]["base"] == "https://ntfy.example.com"
+    assert "topsecret" not in repr(s)

tests/test_service_health_providers.py

@@ -0,0 +1,100 @@
+"""Tests for providers_health — probe logic, status classification, sanitization, and bounded timeout."""
+import pytest
+
+from src import service_health as sh
+
+
+def _raise(*_a, **_k):
+    raise RuntimeError("connection refused")
+
+
+def _ep(name):
+    return {"name": name, "base_url": f"http://{name}:8000/v1", "api_key": "sk-secret"}
+
+
+def test_providers_disabled_without_endpoints():
+    assert sh.providers_health([])["status"] == sh.DISABLED
+
+
+def test_providers_ok_all_reachable():
+    s = sh.providers_health([_ep("a")],
+                            probe=lambda base, key, timeout: ["m1", "m2"])
+    assert s["status"] == sh.OK
+    assert s["meta"]["endpoints"][0]["model_count"] == 2
+
+
+def test_providers_degraded_some_empty():
+    def probe(base, key, timeout):
+        return ["m1"] if "good" in base else []
+
+    s = sh.providers_health([_ep("good"), _ep("bad")], probe=probe)
+    assert s["status"] == sh.DEGRADED
+
+
+def test_providers_down_all_fail():
+    s = sh.providers_health([_ep("a")], probe=_raise)
+    assert s["status"] == sh.DOWN
+
+
+def test_providers_meta_never_leaks_api_key():
+    s = sh.providers_health([_ep("a")],
+                            probe=lambda base, key, timeout: ["m1"])
+    assert "sk-secret" not in repr(s)
+
+
+def test_providers_name_fallback_is_sanitized():
+    # No display name → falls back to the base_url, which must be sanitized.
+    ep = {"base_url": "http://user:k3y@prov.local:9000/v1?api_key=zzz", "api_key": "sk-x"}
+    s = sh.providers_health([ep], probe=lambda b, k, t: ["m1"])
+    entry = s["meta"]["endpoints"][0]
+    assert entry["name"] == "http://prov.local:9000/v1"
+    assert "k3y" not in repr(s) and "zzz" not in repr(s) and "sk-x" not in repr(s)
+
+
+def test_providers_probe_exception_maps_to_category():
+    def boom(base, key, timeout):
+        raise RuntimeError(f"500 from {base} with key {key}")  # would leak base+key
+    s = sh.providers_health([_ep("a")], probe=boom)
+    assert s["status"] == sh.DOWN
+    assert s["meta"]["endpoints"][0]["error"] == "error"
+    assert "sk-secret" not in repr(s) and "http://a" not in repr(s)
+
+
+def test_providers_bounded_marks_slow_as_timeout(monkeypatch):
+    import time
+    monkeypatch.setattr(sh, "_FANOUT_BUDGET", 1)
+
+    def probe(base, key, timeout):
+        if "slow" in base:
+            time.sleep(10)          # would blow the budget if unbounded
+        return ["m1"]
+
+    eps = [{"name": "fast", "base_url": "http://fast", "api_key": "k"},
+           {"name": "slow", "base_url": "http://slow", "api_key": "k"}]
+    t0 = time.monotonic()
+    out = sh.providers_health(eps, probe=probe)
+    elapsed = time.monotonic() - t0
+    assert elapsed < 4, f"providers_health not bounded: took {elapsed:.1f}s"
+    by = {e["name"]: e for e in out["meta"]["endpoints"]}
+    assert by["fast"]["ok"] is True
+    assert by["slow"]["ok"] is False and by["slow"]["error"] == "timeout"
+    assert out["status"] == sh.DEGRADED
+
+
+def test_providers_bounded_with_many_slow_endpoints(monkeypatch):
+    import time
+    monkeypatch.setattr(sh, "_FANOUT_BUDGET", 1)
+
+    def probe(base, key, timeout):
+        time.sleep(10)
+        return ["m1"]
+
+    eps = [{"name": f"ep{i}", "base_url": f"http://ep{i}", "api_key": "k"}
+           for i in range(25)]
+    t0 = time.monotonic()
+    out = sh.providers_health(eps, probe=probe)
+    elapsed = time.monotonic() - t0
+    # 25 endpoints * sleep would be huge if sequential; bounded keeps it ~budget.
+    assert elapsed < 4, f"not bounded with many endpoints: {elapsed:.1f}s"
+    assert out["status"] == sh.DOWN
+    assert all(e["error"] == "timeout" for e in out["meta"]["endpoints"])

tests/test_service_health_search.py

@@ -0,0 +1,79 @@
+"""Tests for searxng_health — probe logic, status classification, and sanitization."""
+import types
+
+import pytest
+
+from src import service_health as sh
+
+
+def _resp(status_code):
+    return types.SimpleNamespace(status_code=status_code)
+
+
+def _raise(*_a, **_k):
+    raise RuntimeError("connection refused")
+
+
+def test_searxng_disabled_when_other_provider():
+    s = sh.searxng_health({"search_provider": "brave"})
+    assert s["status"] == sh.DISABLED
+
+
+def test_searxng_ok_on_healthz():
+    s = sh.searxng_health(
+        {"search_provider": "searxng", "search_url": "http://sx:8080"},
+        http_get=lambda url, timeout: _resp(200),
+    )
+    assert s["status"] == sh.OK
+    assert s["meta"]["probed"] == "/healthz"
+
+
+def test_searxng_ok_on_root_fallback():
+    def getter(url, timeout):
+        return _resp(404) if url.endswith("/healthz") else _resp(200)
+
+    s = sh.searxng_health(
+        {"search_provider": "searxng", "search_url": "http://sx:8080"},
+        http_get=getter,
+    )
+    assert s["status"] == sh.OK
+    assert s["meta"]["probed"] == "/"
+
+
+def test_searxng_down_on_exception():
+    s = sh.searxng_health(
+        {"search_provider": "searxng", "search_url": "http://sx:8080"},
+        http_get=_raise,
+    )
+    assert s["status"] == sh.DOWN
+
+
+def test_searxng_down_on_5xx():
+    s = sh.searxng_health(
+        {"search_provider": "searxng", "search_url": "http://sx:8080"},
+        http_get=lambda url, timeout: _resp(502),
+    )
+    assert s["status"] == sh.DOWN
+
+
+def test_searxng_meta_redacts_instance_url():
+    s = sh.searxng_health(
+        {"search_provider": "searxng",
+         "search_url": "http://user:s3cr3t@searx.local:8080/?token=zzz"},
+        http_get=lambda url, timeout: _resp(200),
+    )
+    blob = repr(s)
+    assert "s3cr3t" not in blob and "zzz" not in blob and "user:" not in blob
+    assert s["meta"]["instance"] == "http://searx.local:8080"
+
+
+def test_searxng_down_uses_error_category_not_raw_exception():
+    def boom(url, timeout):
+        raise RuntimeError("failed connecting to http://user:pw@searx.local secret-token")
+    s = sh.searxng_health(
+        {"search_provider": "searxng", "search_url": "http://searx.local"},
+        http_get=boom,
+    )
+    assert s["status"] == sh.DOWN
+    assert s["meta"]["error"] == "error"           # controlled category token
+    assert "secret-token" not in repr(s) and "pw@" not in repr(s)