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

routes/chat_routes.py

@@ -1327,8 +1327,6 @@ def setup_chat_routes(
                                     "doc_stream_open", "doc_stream_delta",
                                     "doc_update", "doc_suggestions", "ui_control",
                                     "rounds_exhausted",
-                                    "loop_breaker_triggered",
-                                    "intent_nudge_exhausted",
                                     "ask_user",
                                     "plan_update",
                                 ):

src/agent_loop.py

@@ -42,167 +42,6 @@ from src.agent_tools import (
 
 logger = logging.getLogger(__name__)
 
-# Redaction patterns for common secret-bearing shapes. Explicit and tested
-# (see tests/test_loop_guard_signals.py) rather than one clever broad regex —
-# safety first, but we try not to mangle harmless prose. Applied in order.
-_REDACTED = "[redacted]"
-
-# Cookie: ... / Set-Cookie: ... — redact the rest of the line (cookies hold spaces).
-_SENSITIVE_COOKIE_RE = re.compile(
-    r"(?i)\b((?:set-)?cookie\s*[:=]\s*)[^\r\n]+"
-)
-# URL credentials, e.g. postgres://user:pass@host/db. The password half allows
-# inner colons (postgres://user:pa:ss@host/db) but still stops at / and @.
-_SENSITIVE_URL_CRED_RE = re.compile(
-    r"(?i)\b([a-z][a-z0-9+.\-]*://)[^\s:/@]+:[^\s/@]+@"
-)
-# Prefix-only discovery regexes. Each matches the key and its separator (the part
-# we KEEP); the value that follows is found by a linear scanner rather than by a
-# regex, so there is no backtracking-prone quantifier over uncontrolled input.
-#
-# Authorization: Bearer <tok> / Authorization: Basic "two word secret"
-_AUTH_PREFIX_RE = re.compile(
-    r"(?i)authorization\s*[:=]\s*(?:bearer|basic)\s+"
-)
-# Provider-prefixed env names, e.g. OPENAI_API_KEY=..., AWS_SECRET_ACCESS_KEY=...,
-# GITHUB_TOKEN=... — require a sensitive suffix preceded by `_` so benign names
-# that merely end in KEY (MONKEY, TURKEY) are left alone.
-_ENV_PREFIX_RE = re.compile(
-    r"(?:export\s+)?\b[A-Z][A-Z0-9_]*"
-    r"_(?:KEY|TOKEN|SECRET|PASSWORD|PASSWD|PWD|CREDENTIALS?)\s*=\s*"
-)
-# Generic sensitive key, e.g. password=..., api_key: ..., client_secret=...
-_KEY_PREFIX_RE = re.compile(
-    r"(?i)\b(?:password|passwd|pwd|token|api[_-]?key|client_secret|secret)\b\s*[:=]\s*"
-)
-# Obvious provider-shaped bare tokens (no surrounding key needed).
-_SENSITIVE_BARE_TOKEN_RE = re.compile(
-    r"\b("
-    r"sk-[A-Za-z0-9_\-]{16,}"          # OpenAI / Anthropic style
-    r"|gh[pousr]_[A-Za-z0-9]{20,}"     # GitHub PAT
-    r"|xox[baprs]-[A-Za-z0-9\-]{10,}"  # Slack
-    r"|AKIA[0-9A-Z]{16}"               # AWS access key id
-    r"|hf_[A-Za-z0-9]{16,}"            # Hugging Face token
-    r"|AIza[0-9A-Za-z_\-]{20,}"        # Google API key
-    r")\b"
-)
-
-
-def _consume_secret_value_end(text: str, start: int) -> int:
-    """Return the exclusive end index of the secret value beginning at ``start``.
-
-    If the value is quoted, scan to the matching unescaped quote (backslash
-    escapes are skipped two chars at a time). Otherwise scan to the first
-    whitespace, comma, or semicolon. The scan is linear in the length of the
-    input, so it cannot exhibit catastrophic backtracking.
-    """
-    n = len(text)
-    if start >= n:
-        return start
-    quote = text[start]
-    if quote in ("'", '"'):
-        i = start + 1
-        while i < n:
-            ch = text[i]
-            if ch == "\\":
-                i += 2
-                continue
-            if ch == quote:
-                return i + 1
-            i += 1
-        return n  # unterminated quote: redact to the end
-    i = start
-    while i < n and not text[i].isspace() and text[i] not in (",", ";"):
-        i += 1
-    return i
-
-
-def _redact_after_prefix(text: str, prefix_re: "re.Pattern") -> str:
-    """Redact the value following each ``prefix_re`` match using a linear scan."""
-    result = []
-    pos = 0
-    n = len(text)
-    while pos < n:
-        match = prefix_re.search(text, pos)
-        if match is None:
-            result.append(text[pos:])
-            break
-        result.append(text[pos:match.end()])
-        value_end = _consume_secret_value_end(text, match.end())
-        if value_end > match.end():
-            result.append(_REDACTED)
-            pos = value_end
-        else:
-            # Empty value: nothing to redact; step past the prefix and continue.
-            pos = match.end()
-            if pos < n:
-                result.append(text[pos])
-                pos += 1
-    return "".join(result)
-
-
-def _redact_private_keys(text: str) -> str:
-    """Replace PEM private-key blocks with a placeholder via linear scanning.
-
-    Finds ``-----BEGIN `` markers, verifies the header names a PRIVATE KEY,
-    locates the matching ``-----END `` marker, and collapses the whole block.
-    No regex is used, so the (multi-line, uncontrolled) body cannot trigger
-    polynomial matching.
-    """
-    begin_marker = "-----BEGIN "
-    end_marker = "-----END "
-    dash = "-----"
-    max_header = 64  # generous bound on "[TYPE ]PRIVATE KEY"
-    result = []
-    pos = 0
-    while True:
-        begin = text.find(begin_marker, pos)
-        if begin == -1:
-            result.append(text[pos:])
-            return "".join(result)
-        header_start = begin + len(begin_marker)
-        header_close = text.find(dash, header_start)
-        if (
-            header_close == -1
-            or header_close - header_start > max_header
-            or not text[header_start:header_close].endswith("PRIVATE KEY")
-        ):
-            result.append(text[pos:header_start])
-            pos = header_start
-            continue
-        end = text.find(end_marker, header_close)
-        if end == -1:
-            result.append(text[pos:])
-            return "".join(result)
-        end_header_start = end + len(end_marker)
-        end_close = text.find(dash, end_header_start)
-        if (
-            end_close == -1
-            or end_close - end_header_start > max_header
-            or not text[end_header_start:end_close].endswith("PRIVATE KEY")
-        ):
-            result.append(text[pos:header_start])
-            pos = header_start
-            continue
-        result.append(text[pos:begin])
-        result.append("[redacted private key]")
-        pos = end_close + len(dash)
-
-
-def _redact_sensitive_text(value: object) -> str:
-    """Redact obvious credential values before surfacing tool output."""
-    if value is None:
-        return ""
-
-    text = str(value)
-    text = _redact_private_keys(text)
-    text = _redact_after_prefix(text, _AUTH_PREFIX_RE)
-    text = _SENSITIVE_COOKIE_RE.sub(r"\1" + _REDACTED, text)
-    text = _SENSITIVE_URL_CRED_RE.sub(r"\1" + _REDACTED + "@", text)
-    text = _redact_after_prefix(text, _ENV_PREFIX_RE)
-    text = _redact_after_prefix(text, _KEY_PREFIX_RE)
-    return _SENSITIVE_BARE_TOKEN_RE.sub(_REDACTED, text)
-
 
 def _load_mcp_disabled_map() -> Dict[str, set]:
     """Load per-server disabled tool sets from the database."""
@@ -3043,7 +2882,6 @@ async def stream_agent_loop(
     # signatures + consecutive no-text tool rounds to bail early.
     _recent_call_sigs = collections.deque(maxlen=6)
     _stuck_rounds = 0
-    _MAX_STUCK_ROUNDS = 4  # consecutive no-progress rounds before loop-breaker bails
     # Frequency of each exact call signature (tool + args), for the runaway
     # backstop. Counting identical repeats — not distinct same-tool calls —
     # lets a legit batch (e.g. 18 calendar events at once) through.
@@ -3573,22 +3411,17 @@ async def stream_agent_loop(
             # promise: short response (<400 chars), no fenced code/answer,
             # and an action-intent phrase was matched. Long answers that
             # happen to contain "let me know" are not stalls.
-            _promise_shape = (
+            _looks_like_promise = (
                 not guide_only
                 and _intent_match is not None
                 and len(_intent_text) < 400
                 and "```" not in _intent_text
+                and _intent_nudge_count < _MAX_INTENT_NUDGES
             )
-            _looks_like_promise = _promise_shape and _intent_nudge_count < _MAX_INTENT_NUDGES
             if _looks_like_promise:
                 _intent_nudge_count += 1
                 _matched_phrase = _intent_match.group(0).strip()
-                # Don't log the matched phrase — it's raw model text that may
-                # carry credentials. Structural metadata only.
-                logger.info(
-                    "[agent] intent-without-action nudge #%d on round %d",
-                    _intent_nudge_count, round_num,
-                )
+                logger.info(f"[agent] intent-without-action nudge #{_intent_nudge_count} on round {round_num}: {_matched_phrase!r}")
                 _lower_phrase = _matched_phrase.lower()
                 _cookbook_log_hint = ""
                 if any(_word in _lower_phrase for _word in ("log", "logs", "output", "tail", "status")):
@@ -3614,24 +3447,6 @@ async def stream_agent_loop(
                 # Visible signal in the stream so the user knows we caught it.
                 yield f'data: {json.dumps({"type": "agent_step", "round": round_num + 1})}\n\n'
                 continue
-            # The model keeps announcing actions it never takes and we've spent
-            # every nudge — surface why the turn is ending instead of letting it
-            # look like a clean completion.
-            if _promise_shape and _intent_nudge_count >= _MAX_INTENT_NUDGES:
-                _matched_phrase = _intent_match.group(0).strip()
-                _matched_phrase_safe = _redact_sensitive_text(_matched_phrase)
-                _in_message = (
-                    f"Intent-nudge cap reached on round {round_num}: the model "
-                    f"announced an action ({_matched_phrase_safe!r}) without a tool call "
-                    f"after {_intent_nudge_count} nudge(s); ending the turn."
-                )
-                # Do not log the matched phrase, even redacted. It is raw model
-                # text and may contain credentials; keep logs structural only.
-                logger.warning(
-                    "[agent] intent-nudge cap exhausted on round %d (%d/%d)",
-                    round_num, _intent_nudge_count, _MAX_INTENT_NUDGES,
-                )
-                yield f'data: {json.dumps({"type": "intent_nudge_exhausted", "round": round_num, "nudges": _intent_nudge_count, "max_nudges": _MAX_INTENT_NUDGES, "message": _in_message})}\n\n'
             break  # no tools — done
 
         # ── Loop-breaker (Terminus-style stall detector) ──────────────
@@ -3664,23 +3479,10 @@ async def stream_agent_loop(
         # Distinct calls to one tool (a real batch) are legitimate work, so we
         # count identical call signatures, not raw per-tool-type totals.
         _runaway = _detect_runaway_call(_call_freq)
-        if _stuck_rounds >= _MAX_STUCK_ROUNDS or _runaway:
+        if _stuck_rounds >= 4 or _runaway:
             reason = (f"calling {_runaway} with identical arguments over and over" if _runaway
                       else "repeating the same tool calls without new progress")
-            _lb_message = (
-                f"Loop-breaker stopped the agent on round {round_num}: {reason}. "
-                "Forced one tool-free round to converge on an answer or state what's blocked."
-            )
-            # Log structural metadata only — `_sig` is raw tool-call content
-            # that may carry credentials.
-            logger.warning(
-                "[agent] loop-breaker tripped on round %d (%s); "
-                "stuck_rounds=%d/%d runaway=%r",
-                round_num, reason, _stuck_rounds, _MAX_STUCK_ROUNDS, _runaway,
-            )
-            # Surface the stop cause to the stream so the user (and journalctl)
-            # can tell a guard fired, not a clean completion.
-            yield f'data: {json.dumps({"type": "loop_breaker_triggered", "round": round_num, "reason": reason, "stuck_rounds": _stuck_rounds, "max_stuck_rounds": _MAX_STUCK_ROUNDS, "runaway": _runaway, "message": _lb_message})}\n\n'
+            logger.warning(f"[agent] loop-breaker tripped on round {round_num} ({reason}); sig={_sig[:80]!r}")
             # The model has been executing tools, so its results are already
             # in context. Force ONE tool-free round to converge: write the
             # answer from what it has, or state plainly what's blocking it.
@@ -3760,10 +3562,6 @@ async def stream_agent_loop(
                 cmd_display = block.content.split("\n")[0].strip()[:80]
             else:
                 cmd_display = full_command
-            # The display string is streamed (tool_start/tool_output) and persisted;
-            # redact any secrets in it. The executed block content is left untouched
-            # so tool execution still sees the real command.
-            cmd_display = _redact_sensitive_text(cmd_display)
 
             if tool_policy and tool_policy.blocks(block.tool_type):
                 desc = f"{block.tool_type}: BLOCKED"
@@ -3809,15 +3607,8 @@ async def stream_agent_loop(
                     evt = await _progress_q.get()
                     if evt is None:
                         break
-                    # Redact secrets in the live tail before streaming — the
-                    # final tool_output is redacted, so the progress tail must
-                    # be too, or a secret could flash by mid-run. Copy so we
-                    # don't mutate the tool's own event payload.
-                    _evt = dict(evt)
-                    if isinstance(_evt.get("tail"), str):
-                        _evt["tail"] = _redact_sensitive_text(_evt["tail"])
                     yield (
-                        f'data: {json.dumps({"type": "tool_progress", "tool": block.tool_type, "round": round_num, **_evt})}\n\n'
+                        f'data: {json.dumps({"type": "tool_progress", "tool": block.tool_type, "round": round_num, **evt})}\n\n'
                     )
                 desc, result = await _tool_task
 
@@ -3883,7 +3674,7 @@ async def stream_agent_loop(
                                 result["results"] = _clean
                             elif "stdout" in result:
                                 result["stdout"] = _clean
-                        except Exception:
+                        except (json.JSONDecodeError, Exception):
                             pass
 
             # Emit doc-specific event for document tools — the frontend
@@ -3953,29 +3744,29 @@ async def stream_agent_loop(
                 # empty) stdout/stderr; fall back to the error so the "timed
                 # out" reason reaches the UI instead of a blank result.
                 raw = result["stdout"] or result["stderr"] or result.get("error", "")
-                output_text = _truncate(_redact_sensitive_text(raw))
+                output_text = _truncate(raw)
             elif "output" in result:
                 # bash / python canonical result: {"output": ..., "exit_code": ...}
                 raw = result["output"] or ""
-                output_text = _truncate(_redact_sensitive_text(raw))
+                output_text = _truncate(raw)
             elif "response" in result:
                 # AI interaction tools (chat_with_model, send_to_session)
                 label = result.get("model", result.get("session_name", "AI"))
-                output_text = _truncate(_redact_sensitive_text(f"{label}: {result['response']}"))
+                output_text = _truncate(f"{label}: {result['response']}")
             elif "content" in result:
-                output_text = _truncate(_redact_sensitive_text(result["content"]))
+                output_text = _truncate(result["content"])
             elif "results" in result:
-                output_text = _truncate(_redact_sensitive_text(result["results"]))
+                output_text = _truncate(result["results"])
             elif "session_id" in result and "name" in result:
                 output_text = f"Session created: {result['name']} (id: {result['session_id']})"
             elif "success" in result:
                 output_text = (
                     f"Written: {result.get('path', '')}"
                     if result["success"]
-                    else f"Error: {_redact_sensitive_text(result.get('error', ''))}"
+                    else f"Error: {result.get('error', '')}"
                 )
             elif "error" in result:
-                output_text = _truncate(_redact_sensitive_text(result["error"]))
+                output_text = _truncate(result["error"])
 
             # Emit tool_output (include ui_event data if present)
             tool_output_data = {"type": "tool_output", "tool": block.tool_type, "command": cmd_display, "output": output_text, "exit_code": result.get("exit_code")}

static/js/chat.js

@@ -2201,23 +2201,6 @@ import { wireArrowUpRecall, getLastUserMessageFromChatHistory } from './composer
                   _chatBox.appendChild(note);
                   try { note.scrollIntoView({ block: 'end', behavior: 'smooth' }); } catch (_) { uiModule.scrollHistory && uiModule.scrollHistory(); }
                 }
-              } else if (json.type === 'loop_breaker_triggered' || json.type === 'intent_nudge_exhausted') {
-                // A loop guard ended the turn — surface why so it isn't mistaken
-                // for a clean completion or a silent stall.
-                const _chatBox = document.getElementById('chat-history');
-                if (!_isBg && _chatBox) {
-                  const note = document.createElement('div');
-                  note.className = 'stopped-indicator loop-guard-stop';
-                  const label = document.createElement('span');
-                  label.className = 'rounds-exhausted-label';
-                  label.textContent = json.message ||
-                    (json.type === 'loop_breaker_triggered'
-                      ? 'Stopped by the loop-breaker (no new progress).'
-                      : 'Stopped: announced an action but never called the tool.');
-                  note.appendChild(label);
-                  _chatBox.appendChild(note);
-                  try { note.scrollIntoView({ block: 'end', behavior: 'smooth' }); } catch (_) { uiModule.scrollHistory && uiModule.scrollHistory(); }
-                }
               } else if (json.type === 'model_actual') {
                 if (!_isBg && holder) {
                   holder._requestedModel = json.requested_model || holder._requestedModel || modelName;

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_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_loop_guard_signals.py

@@ -1,350 +0,0 @@
-"""Regression: stream_agent_loop surfaces *why* a guard ended the turn.
-
-Two internal guards used to stop the agent in ways that looked like a clean
-completion or a vague blocked message:
-
-  * the loop-breaker stall detector -> now emits `loop_breaker_triggered`
-  * the intent-without-action nudge cap -> now emits `intent_nudge_exhausted`
-
-These tests run the real loop body against a fake LLM stream (no model calls,
-no sleeps) and assert the structured stop event is emitted.
-"""
-
-import asyncio
-import json
-import logging
-
-import pytest
-
-import src.agent_loop as al
-
-
-def _collect(gen):
-    async def _run():
-        return [c async for c in gen]
-    return asyncio.run(_run())
-
-
-def _types(chunks):
-    out = []
-    for c in chunks:
-        if c.startswith("data: ") and not c.startswith("data: [DONE]"):
-            try:
-                out.append(json.loads(c[6:]))
-            except Exception:
-                pass
-    return out
-
-
-def _patch_common(monkeypatch):
-    monkeypatch.setattr(al, "get_setting", lambda key, default=None: default, raising=False)
-    monkeypatch.setattr(al, "get_mcp_manager", lambda: None, raising=False)
-    monkeypatch.setattr(al, "estimate_tokens", lambda *a, **k: 10, raising=False)
-
-    async def _fake_exec(block, *a, **k):
-        return ("bash", {"output": "ok", "exit_code": 0})
-    monkeypatch.setattr(al, "execute_tool_block", _fake_exec, raising=False)
-
-
-def _run_loop(monkeypatch, round_text, max_rounds, relevant_tools={"bash"}):
-    async def _fake_stream(_candidates, messages, **kwargs):
-        yield f'data: {json.dumps({"delta": round_text})}\n\n'
-        yield "data: [DONE]\n\n"
-    monkeypatch.setattr(al, "stream_llm_with_fallback", _fake_stream, raising=False)
-
-    gen = al.stream_agent_loop(
-        "http://x/v1", "m",
-        [{"role": "user", "content": "do a long multi-step task"}],
-        max_rounds=max_rounds,
-        relevant_tools=relevant_tools,
-    )
-    return _types(_collect(gen))
-
-
-def test_emits_loop_breaker_triggered_on_repeated_no_progress(monkeypatch):
-    _patch_common(monkeypatch)
-    # Same exact tool call every round, no answer text -> stuck-round streak
-    # trips the loop-breaker once the cap is reached.
-    events = _run_loop(monkeypatch, "```bash\necho hi\n```", max_rounds=8)
-    lb = [e for e in events if e.get("type") == "loop_breaker_triggered"]
-    assert lb, events
-    e = lb[0]
-    assert e["reason"]
-    assert e["max_stuck_rounds"] == 4
-    assert e["stuck_rounds"] >= 4
-    assert "message" in e
-
-
-def test_no_loop_breaker_on_normal_finish(monkeypatch):
-    _patch_common(monkeypatch)
-    events = _run_loop(monkeypatch, "All done, here is your answer.", max_rounds=8)
-    assert not any(e.get("type") == "loop_breaker_triggered" for e in events), events
-
-
-def test_emits_intent_nudge_exhausted_when_cap_reached(monkeypatch):
-    _patch_common(monkeypatch)
-    # The model keeps announcing an action with no tool call. After the nudge
-    # cap is spent, the turn ends with an explicit intent_nudge_exhausted event.
-    events = _run_loop(monkeypatch, "Let me check the logs now", max_rounds=5)
-    inx = [e for e in events if e.get("type") == "intent_nudge_exhausted"]
-    assert inx, events
-    e = inx[0]
-    assert e["max_nudges"] == 2
-    assert e["nudges"] >= 2
-    assert "message" in e
-
-
-def test_no_intent_nudge_exhausted_on_normal_finish(monkeypatch):
-    _patch_common(monkeypatch)
-    events = _run_loop(monkeypatch, "Here is the complete answer to your question.", max_rounds=5)
-    assert not any(e.get("type") == "intent_nudge_exhausted" for e in events), events
-
-
-def _assert_guard_log_safe(caplog, *, structural, secret="secret123"):
-    """The guard's own structural log line fired, and that record carries no raw
-    secret. Scoped to the guard's records on purpose: an unrelated, pre-existing
-    round-summary log echoes raw model text and is out of scope for this PR."""
-    records = [r for r in caplog.records if structural in r.getMessage()]
-    assert records, caplog.text
-    for r in records:
-        assert secret not in r.getMessage(), r.getMessage()
-
-
-def test_intent_nudge_logging_does_not_leak_secret(monkeypatch, caplog):
-    # The model announces an action (no tool call) with a secret in the text.
-    # The nudge logger must record only structural metadata, never the matched
-    # phrase — so the credential never lands in journalctl.
-    _patch_common(monkeypatch)
-    with caplog.at_level(logging.INFO, logger="src.agent_loop"):
-        events = _run_loop(monkeypatch, "Let me check api_key=secret123 now", max_rounds=5)
-    assert any(e.get("type") == "intent_nudge_exhausted" for e in events), events
-    _assert_guard_log_safe(caplog, structural="intent-without-action nudge")
-
-
-def test_loop_breaker_logging_does_not_leak_secret(monkeypatch, caplog):
-    # A repeated tool command carrying a secret trips the loop-breaker. The
-    # structural log must not contain `_sig` / raw tool-call content.
-    _patch_common(monkeypatch)
-    with caplog.at_level(logging.INFO, logger="src.agent_loop"):
-        events = _run_loop(monkeypatch, "```bash\necho api_key=secret123\n```", max_rounds=8)
-    assert any(e.get("type") == "loop_breaker_triggered" for e in events), events
-    _assert_guard_log_safe(caplog, structural="loop-breaker tripped")
-
-
-def test_redacts_sensitive_tool_output_before_surfacing():
-    text = al._redact_sensitive_text(
-        "password: private-value\n"
-        "api_key=private-key\n"
-        "Authorization: Bearer private-token\n"
-        "normal output"
-    )
-
-    assert "private-value" not in text
-    assert "private-key" not in text
-    assert "private-token" not in text
-    assert "password: [redacted]" in text
-    assert "api_key=[redacted]" in text
-    assert "Authorization: Bearer [redacted]" in text
-    assert "normal output" in text
-
-
-_GCP_API_KEY_SAMPLE = "AI" + "za" + ("A" * 35)
-
-# (input, secret substring that must be gone, expected substring that must remain)
-_REDACTION_CASES = [
-    ("Authorization: Bearer abc123tok", "abc123tok", "Authorization: Bearer [redacted]"),
-    ("Authorization: Basic dXNlcjpwYXNz", "dXNlcjpwYXNz", "Authorization: Basic [redacted]"),
-    # Quoted Authorization value (spaces) must be redacted whole.
-    ('Authorization: Bearer "two word secret"', "two word secret", "Authorization: Bearer [redacted]"),
-    # Escaped quote inside a quoted secret must not leak the tail.
-    (r'password="abc\"def secret"', "def secret", "password=[redacted]"),
-    # URL password containing a colon must still be redacted whole.
-    ("postgres://user:pa:ss@host/db", "pa:ss", "postgres://[redacted]@host/db"),
-    # Provider-shaped bare tokens.
-    ("token is hf_abcdefghij1234567890XYZ", "hf_abcdefghij1234567890XYZ", "[redacted]"),
-    ("key " + _GCP_API_KEY_SAMPLE, _GCP_API_KEY_SAMPLE, "[redacted]"),
-    ("Cookie: session=abc123secret", "abc123secret", "Cookie: [redacted]"),
-    ("Set-Cookie: sid=xyz789; HttpOnly", "xyz789", "Set-Cookie: [redacted]"),
-    ("postgres://user:pa55word@host/db", "pa55word", "postgres://[redacted]@host/db"),
-    ("client_secret=supersecretvalue", "supersecretvalue", "client_secret=[redacted]"),
-    ("OPENAI_API_KEY=abcd1234deadbeef", "abcd1234deadbeef", "OPENAI_API_KEY=[redacted]"),
-    # Quoted multi-word env value must be fully redacted, not clipped at the space.
-    ('OPENAI_API_KEY="two word secret"', "two word secret", "OPENAI_API_KEY=[redacted]"),
-    ('password: "my secret value"', "my secret value", "password: [redacted]"),
-    ("here is sk-abcdefghij1234567890", "sk-abcdefghij1234567890", "[redacted]"),
-    (
-        "-----BEGIN PRIVATE KEY-----\nMIIfakeKEYbody\n-----END PRIVATE KEY-----",
-        "MIIfakeKEYbody",
-        "[redacted private key]",
-    ),
-]
-
-
-@pytest.mark.parametrize("raw, secret, expected", _REDACTION_CASES)
-def test_redaction_covers_requested_secret_shapes(raw, secret, expected):
-    out = al._redact_sensitive_text(raw)
-    assert secret not in out, out
-    assert expected in out, out
-
-
-@pytest.mark.parametrize("raw", [
-    "the build completed in 3.2s with 0 errors",
-    "password reset email sent to the user",
-    "Listing 5 files: a.py b.py c.py d.py e.py",
-    "https://example.com/path?page=2",
-    # Benign uppercase names that merely end in KEY must not be redacted.
-    "MONKEY=banana",
-    "TURKEY=dinner",
-])
-def test_redaction_keeps_normal_output_readable(raw):
-    assert al._redact_sensitive_text(raw) == raw
-
-
-def test_redacts_before_truncating():
-    # A secret near the start must be gone even if truncation would otherwise
-    # only clip the tail — redaction runs first.
-    raw = "api_key=topsecretvalue " + ("x" * 50_000)
-    out = al._truncate(al._redact_sensitive_text(raw))
-    assert "topsecretvalue" not in out
-    assert "api_key=[redacted]" in out
-
-
-def _run_tool_result(monkeypatch, tool, exec_result, max_rounds=2):
-    """Drive one tool round whose execution returns `exec_result`, and collect
-    the streamed events. Used to assert restored per-tool-result emissions."""
-    _patch_common(monkeypatch)
-
-    async def _fake_exec(block, *a, **k):
-        return (tool, exec_result)
-    monkeypatch.setattr(al, "execute_tool_block", _fake_exec, raising=False)
-
-    round_text = f"```{tool}\n{{}}\n```"
-
-    async def _fake_stream(_candidates, messages, **kwargs):
-        yield f'data: {json.dumps({"delta": round_text})}\n\n'
-        yield "data: [DONE]\n\n"
-    monkeypatch.setattr(al, "stream_llm_with_fallback", _fake_stream, raising=False)
-
-    gen = al.stream_agent_loop(
-        "http://x/v1", "m",
-        [{"role": "user", "content": "do something"}],
-        max_rounds=max_rounds,
-        relevant_tools={tool},
-    )
-    return _types(_collect(gen))
-
-
-def test_restores_doc_suggestions_event(monkeypatch):
-    events = _run_tool_result(
-        monkeypatch, "suggest_document",
-        {"action": "suggest", "doc_id": "d1", "suggestions": [{"text": "x"}], "exit_code": 0},
-    )
-    assert any(e.get("type") == "doc_suggestions" for e in events), events
-
-
-def test_restores_doc_update_event(monkeypatch):
-    events = _run_tool_result(
-        monkeypatch, "edit_document",
-        {"action": "edit", "doc_id": "d1", "content": "body", "version": 2,
-         "title": "T", "language": "md", "exit_code": 0},
-    )
-    # A native document block also emits doc_update AFTER tool_output, so a plain
-    # "any doc_update" check would pass even if the restored generic block were
-    # gone. Prove the restored block fires BEFORE the first tool_output.
-    types = [e.get("type") for e in events]
-    assert "doc_update" in types, events
-    assert "tool_output" in types, events
-    assert types.index("doc_update") < types.index("tool_output"), types
-
-
-def test_restores_ui_control_event(monkeypatch):
-    events = _run_tool_result(
-        monkeypatch, "ui_control",
-        {"ui_event": "toggle", "toggle_name": "bash", "state": "off", "exit_code": 0},
-    )
-    assert any(e.get("type") == "ui_control" for e in events), events
-
-
-def test_restores_plan_update_event(monkeypatch):
-    events = _run_tool_result(
-        monkeypatch, "update_plan",
-        {"plan_update": {"steps": [{"text": "step", "done": True}]}, "exit_code": 0},
-    )
-    assert any(e.get("type") == "plan_update" for e in events), events
-
-
-def test_restores_ask_user_event_and_persists_question(monkeypatch):
-    events = _run_tool_result(
-        monkeypatch, "ask_user",
-        {"ask_user": {"question": "Which option?", "options": [{"label": "A"}, {"label": "B"}]},
-         "exit_code": 0},
-    )
-    # Exactly one ask_user event — not re-emitted on a follow-up round.
-    _ask_events = [e for e in events if e.get("type") == "ask_user"]
-    assert len(_ask_events) == 1, events
-    # The question is streamed as assistant text so it persists for replay.
-    # Upstream prepends "\n\n" when full_response already holds streamed text,
-    # so match on containment — and it must be streamed exactly once.
-    _q_deltas = [e for e in events if "Which option?" in (e.get("delta") or "")]
-    assert len(_q_deltas) == 1, events
-    # Setting `_awaiting_user` breaks the loop, so the turn does NOT advance into
-    # another agent round (which would emit an agent_step event) after the ask.
-    assert not any(e.get("type") == "agent_step" for e in events), events
-
-
-def test_redacts_command_display_in_streamed_events(monkeypatch):
-    # A tool command line can carry a secret. The streamed command display
-    # (tool_start / tool_output) must be redacted, even though the real command
-    # passed to execution is left untouched.
-    _patch_common(monkeypatch)
-
-    round_text = "```bash\necho api_key=secret123\n```"
-
-    async def _fake_stream(_candidates, messages, **kwargs):
-        yield f'data: {json.dumps({"delta": round_text})}\n\n'
-        yield "data: [DONE]\n\n"
-    monkeypatch.setattr(al, "stream_llm_with_fallback", _fake_stream, raising=False)
-
-    gen = al.stream_agent_loop(
-        "http://x/v1", "m",
-        [{"role": "user", "content": "run it"}],
-        max_rounds=2,
-        relevant_tools={"bash"},
-    )
-    events = _types(_collect(gen))
-    cmds = [e for e in events if e.get("type") in ("tool_start", "tool_output")]
-    assert cmds, events
-    assert all("secret123" not in (e.get("command") or "") for e in cmds), cmds
-    assert any("api_key=[redacted]" in (e.get("command") or "") for e in cmds), cmds
-
-
-def test_redacts_live_tool_progress_tail(monkeypatch):
-    # A secret in the live progress tail must be redacted before streaming —
-    # otherwise it flashes by before the (already redacted) final tool_output.
-    _patch_common(monkeypatch)
-
-    async def _fake_exec(block, *a, **k):
-        await k["progress_cb"]({"tail": "api_key=secret123", "elapsed_s": 1})
-        return ("bash", {"output": "done", "exit_code": 0})
-    monkeypatch.setattr(al, "execute_tool_block", _fake_exec, raising=False)
-
-    round_text = "```bash\necho hi\n```"
-
-    async def _fake_stream(_candidates, messages, **kwargs):
-        yield f'data: {json.dumps({"delta": round_text})}\n\n'
-        yield "data: [DONE]\n\n"
-    monkeypatch.setattr(al, "stream_llm_with_fallback", _fake_stream, raising=False)
-
-    gen = al.stream_agent_loop(
-        "http://x/v1", "m",
-        [{"role": "user", "content": "run it"}],
-        max_rounds=2,
-        relevant_tools={"bash"},
-    )
-    events = _types(_collect(gen))
-    prog = [e for e in events if e.get("type") == "tool_progress"]
-    assert prog, events
-    assert all("secret123" not in (e.get("tail") or "") for e in prog), prog
-    assert any("api_key=[redacted]" in (e.get("tail") or "") for e in prog), prog
-    # Other fields are preserved.
-    assert any(e.get("elapsed_s") == 1 for e in prog), prog

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)