ops: add Gitea consistency sentinel timer unit

docs/security/user-data-isolation-redline.md

@@ -1,193 +0,0 @@
-# User Data Isolation Redline
-
-Status: mandatory
-
-Decision date: 2026-07-10
-
-Baseline at decision: V1R4C3
-
-## Redline
-
-User data isolation is a hard system-design requirement.
-
-No component may read, write, search, rank, summarize, cache, or report data from another user, tenant, project, session, workspace, branch, agent, or task context unless explicit authorization and context binding exist.
-
-This applies to memory search, filesystem logs, vector indexes, search gateways, autonomous controller state, task records, audit logs, artifacts, and any fallback provider.
-
-## Incident Class
-
-Any cross-context data contamination is a security and reliability incident, not a normal feature bug.
-
-Examples:
-
-- memory search returns records from another project;
-- shared workspace logs are used as current-session evidence;
-- global `memory/` state mixes unrelated users or tasks;
-- fallback provider uses unscoped records;
-- agent reports status based on stale or unrelated context;
-- shared cache leaks content across users, groups, or sessions.
-
-## Core Rule
-
-A query may only access records satisfying all required scope dimensions:
-
-```text
-user_id
-  AND tenant_id / group_id
-  AND session_id or conversation_id
-  AND project_id
-  AND workspace_id
-  AND branch or baseline scope
-  AND agent_id / tool_context when applicable
-```
-
-If a dimension is unavailable, the safe behavior is deny-by-default or text-only fallback.
-
-## Forbidden Design
-
-The following are forbidden in production paths:
-
-1. Global shared memory directories without per-scope namespace.
-2. Search over untagged records.
-3. Search over records whose owner/scope cannot be proven.
-4. Trusting filesystem location alone as authorization.
-5. Treating memory search results as authoritative without provenance.
-6. Cross-session fallback without explicit allowlist.
-7. Auto-reporting from memory results that are not bound to the current runtime/baseline/task.
-8. Using stale context from unrelated branches or projects as current-state evidence.
-
-## Required Design
-
-Every persisted memory/state record must include immutable metadata:
-
-```text
-record_id
-user_id
-tenant_id or group_id
-session_id / conversation_id
-project_id
-workspace_id
-branch
-baseline
-agent_id
-source_tool
-created_at
-schema_version
-sensitivity
-retention_class
-```
-
-Every read/search request must include a scope envelope:
-
-```text
-request_user_id
-request_tenant_id / group_id
-request_session_id / conversation_id
-request_project_id
-request_workspace_id
-allowed_branches / current_branch
-allowed_baselines / current_baseline
-purpose
-```
-
-A record is readable only if its metadata matches the request scope policy.
-
-## Enforcement Layers
-
-### 1. Storage namespace isolation
-
-Memory/state paths must be namespaced:
-
-```text
-memory/{tenant_id}/{user_id}/{project_id}/{session_id}/...
-state/{tenant_id}/{project_id}/{workspace_id}/...
-```
-
-No production component may write unrelated contexts into a single global `memory/` directory.
-
-### 2. Metadata enforcement
-
-Untagged records are unreadable by default.
-
-Records without owner/scope metadata may be archived for forensic review but must not be searched or used for live reasoning.
-
-### 3. Query-time ACL
-
-Search must apply a mandatory scope filter before ranking or summarization.
-
-The ranking stage must never receive out-of-scope records.
-
-### 4. Provenance requirement
-
-Every memory/search result used for operational reporting must carry provenance:
-
-```text
-record_id
-scope metadata
-source path or source index
-match reason
-created_at
-```
-
-Responses must prefer current runtime truth sources over memory:
-
-```text
-getRuntimeStatus
-getBaseline
-listPatches
-explicit user-provided current context
-current branch files
-```
-
-Memory search is advisory only after scope validation.
-
-### 5. Fallback safety
-
-If provider search cannot prove scope isolation, the system must use text-only fallback or no-memory fallback.
-
-Fallback must not silently broaden the scope.
-
-### 6. Write-path isolation
-
-Writers must not append current-session records into shared directories.
-
-Writers must derive the target path from the scope envelope, not from a global default.
-
-### 7. Audit and detection
-
-The system must produce audit events when:
-
-- a record is denied due to scope mismatch;
-- untagged memory is encountered;
-- fallback disables memory search;
-- a provider tries to return out-of-scope records;
-- cross-context contamination is detected.
-
-## Acceptance Criteria
-
-A memory/search provider is allowed only if all tests pass:
-
-1. Same user, same project, same session -> records are visible.
-2. Same user, same project, different session -> visible only if policy permits.
-3. Same user, different project -> denied by default.
-4. Different user, same project -> denied unless explicitly shared by group policy.
-5. Different tenant/group -> denied.
-6. Untagged record -> denied.
-7. Stale branch/baseline record -> not used as current evidence.
-8. Provider fallback without scope support -> memory disabled.
-9. Search result provenance is returned and logged.
-10. Out-of-scope records are filtered before ranking/summarization.
-
-## Runtime Rule
-
-If memory search is enabled without these controls, recurrence is possible.
-
-Therefore, memory search must remain disabled or in scoped allowlist mode until the isolation acceptance suite is green.
-
-## Owner Gate
-
-Any change that enables shared memory search, cross-session recall, global vector indexes, or cross-agent memory reuse requires owner/admin approval.
-
-## Final Rule
-
-User data isolation is a redline. Reliability, convenience, or recall quality may not override it.

ops/systemd/gitea-branch-consistency-sentinel.timer

@@ -0,0 +1,11 @@
+[Unit]
+Description=Run Gitea branch consistency sentinel periodically
+
+[Timer]
+OnBootSec=2min
+OnUnitActiveSec=1min
+RandomizedDelaySec=10s
+Persistent=true
+
+[Install]
+WantedBy=timers.target

scripts/reliability/autonomous_safe_mode.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Safe-mode gate for autonomous mutations.
+
+Controllers should run `check` before writes. Exit code 75 means the
+controller must stop mutation and wait for controlled recovery.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime
+import json
+import os
+from pathlib import Path
+import sys
+
+DEFAULT = "/data/aios-gitops/runtime/state/autonomous-safe-mode.json"
+
+
+def now():
+    return datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def load(path):
+    if not path.exists():
+        return None
+    with path.open("r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--safe-mode-file", default=os.environ.get("AUTONOMOUS_SAFE_MODE_FILE", DEFAULT))
+    parser.add_argument("command", choices=["check", "enter", "clear"])
+    parser.add_argument("--reason", default="")
+    args = parser.parse_args()
+
+    path = Path(args.safe_mode_file)
+
+    if args.command == "check":
+        marker = load(path)
+        if marker:
+            print(json.dumps(marker, ensure_ascii=False, indent=2))
+            return 75
+        return 0
+
+    if args.command == "enter":
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps({
+            "status": "active",
+            "entered_at": now(),
+            "reason": args.reason or "manual",
+        }, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+        return 0
+
+    if args.command == "clear":
+        if path.exists():
+            path.unlink()
+        return 0
+
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/reliability/gitea_branch_consistency_sentinel.py

@@ -0,0 +1,414 @@
+#!/usr/bin/env python3
+"""
+Gitea branch metadata consistency sentinel.
+
+Compares three views of a repository's branch state:
+1. Bare Git refs under refs/heads
+2. Gitea SQLite branch metadata rows with is_deleted=0
+3. Gitea branch API pagination results
+
+On drift or check failure, optionally writes an autonomous safe-mode marker so
+controllers can stop mutating state before the incident expands.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime as _dt
+import json
+import os
+from pathlib import Path
+import shlex
+import sqlite3
+import subprocess
+import sys
+import tempfile
+from typing import Any
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+DEFAULT_REPO_DIR = "/data/aios-gitops/runtime/gitea-bin/data/repositories/aios/aios.git"
+DEFAULT_DB_PATH = "/data/aios-gitops/runtime/gitea-bin/data/gitea.db"
+DEFAULT_API_URL = "http://127.0.0.1:3000/api/v1/repos/aios/aios/branches"
+DEFAULT_STATE_DIR = "/data/aios-gitops/runtime/state"
+DEFAULT_SAFE_MODE_FILE = f"{DEFAULT_STATE_DIR}/autonomous-safe-mode.json"
+DEFAULT_REPORT_FILE = f"{DEFAULT_STATE_DIR}/gitea-branch-consistency-report.json"
+DEFAULT_DOCTOR_CMD = (
+    "/data/aios-gitops/runtime/bin/gitea "
+    "-C /data/aios-gitops/runtime/gitea-bin/custom "
+    "doctor check --run check-db-consistency"
+)
+
+
+def utc_now() -> str:
+    return _dt.datetime.now(_dt.timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def env_bool(name: str, default: bool = False) -> bool:
+    value = os.environ.get(name)
+    if value is None or value == "":
+        return default
+    return value.strip().lower() in {"1", "true", "yes", "y", "on"}
+
+
+def atomic_write_json(path: Path, payload: dict[str, Any], mode: int = 0o644) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent))
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True)
+            handle.write("\n")
+        os.chmod(tmp_name, mode)
+        os.replace(tmp_name, path)
+    finally:
+        try:
+            os.unlink(tmp_name)
+        except FileNotFoundError:
+            pass
+
+
+def run_command(command: list[str], timeout_seconds: int) -> dict[str, Any]:
+    try:
+        proc = subprocess.run(
+            command,
+            check=False,
+            text=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            timeout=timeout_seconds,
+        )
+        return {
+            "ok": proc.returncode == 0,
+            "code": proc.returncode,
+            "stdout": proc.stdout.strip(),
+            "stderr": proc.stderr.strip(),
+            "cmd": command,
+        }
+    except FileNotFoundError as exc:
+        return {"ok": False, "code": 127, "stdout": "", "stderr": str(exc), "cmd": command}
+    except subprocess.TimeoutExpired as exc:
+        return {
+            "ok": False,
+            "code": 124,
+            "stdout": (exc.stdout or "").strip() if isinstance(exc.stdout, str) else "",
+            "stderr": (exc.stderr or "").strip() if isinstance(exc.stderr, str) else "",
+            "cmd": command,
+            "timeout_seconds": timeout_seconds,
+        }
+
+
+def git_branch_names(repo_dir: str, timeout_seconds: int) -> tuple[set[str], dict[str, Any]]:
+    command = ["git", "--git-dir", repo_dir, "for-each-ref", "--format=%(refname:short)", "refs/heads"]
+    result = run_command(command, timeout_seconds)
+    if not result["ok"]:
+        raise RuntimeError(f"git branch ref scan failed: {result['stderr'] or result['stdout']}")
+    names = {line.strip() for line in result["stdout"].splitlines() if line.strip()}
+    result["count"] = len(names)
+    return names, result
+
+
+def sqlite_branch_names(db_path: str, repo_id: int) -> tuple[set[str], dict[str, Any]]:
+    path = Path(db_path)
+    if not path.exists():
+        raise RuntimeError(f"Gitea database does not exist: {db_path}")
+
+    uri = f"file:{urllib.parse.quote(str(path.resolve()))}?mode=ro"
+    try:
+        conn = sqlite3.connect(uri, uri=True, timeout=5)
+        try:
+            cursor = conn.execute(
+                "SELECT name FROM branch WHERE repo_id = ? AND is_deleted = 0 ORDER BY name",
+                (repo_id,),
+            )
+            names = {row[0] for row in cursor.fetchall()}
+        finally:
+            conn.close()
+    except sqlite3.Error as exc:
+        raise RuntimeError(f"SQLite branch metadata query failed: {exc}") from exc
+
+    return names, {"ok": True, "repo_id": repo_id, "count": len(names), "db_path": db_path}
+
+
+def add_query_params(url: str, params: dict[str, Any]) -> str:
+    split = urllib.parse.urlsplit(url)
+    query = dict(urllib.parse.parse_qsl(split.query, keep_blank_values=True))
+    query.update({key: str(value) for key, value in params.items()})
+    return urllib.parse.urlunsplit(
+        (split.scheme, split.netloc, split.path, urllib.parse.urlencode(query), split.fragment)
+    )
+
+
+def read_api_token() -> str | None:
+    token = os.environ.get("GITEA_API_TOKEN")
+    if token:
+        return token.strip()
+
+    token_file = os.environ.get("GITEA_API_TOKEN_FILE")
+    if token_file:
+        token_path = Path(token_file)
+        if token_path.exists():
+            return token_path.read_text(encoding="utf-8").strip()
+
+    return None
+
+
+def api_branch_names(
+    api_url: str,
+    limit: int,
+    timeout_seconds: int,
+    max_pages: int,
+) -> tuple[set[str], dict[str, Any]]:
+    token = read_api_token()
+    names: list[str] = []
+    pages: list[dict[str, Any]] = []
+
+    for page in range(1, max_pages + 1):
+        url = add_query_params(api_url, {"page": page, "limit": limit})
+        headers = {"Accept": "application/json"}
+        if token:
+            headers["Authorization"] = f"token {token}"
+
+        request = urllib.request.Request(url, headers=headers, method="GET")
+        try:
+            with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
+                body = response.read()
+                status = getattr(response, "status", response.getcode())
+        except urllib.error.HTTPError as exc:
+            detail = exc.read().decode("utf-8", errors="replace")
+            raise RuntimeError(f"Gitea branch API HTTP {exc.code}: {detail[:500]}") from exc
+        except urllib.error.URLError as exc:
+            raise RuntimeError(f"Gitea branch API request failed: {exc}") from exc
+        except TimeoutError as exc:
+            raise RuntimeError(f"Gitea branch API request timed out: {exc}") from exc
+
+        try:
+            payload = json.loads(body.decode("utf-8"))
+        except json.JSONDecodeError as exc:
+            raise RuntimeError(f"Gitea branch API returned invalid JSON on page {page}") from exc
+
+        if not isinstance(payload, list):
+            raise RuntimeError(f"Gitea branch API returned non-list payload on page {page}: {type(payload).__name__}")
+
+        page_names = []
+        for item in payload:
+            if not isinstance(item, dict) or "name" not in item:
+                raise RuntimeError(f"Gitea branch API returned malformed branch item on page {page}: {item!r}")
+            page_names.append(str(item["name"]))
+
+        names.extend(page_names)
+        pages.append({"page": page, "status": status, "count": len(page_names)})
+
+        if len(page_names) < limit:
+            break
+    else:
+        raise RuntimeError(f"Gitea branch API exceeded max_pages={max_pages}; refusing to trust partial result")
+
+    return set(names), {
+        "ok": True,
+        "api_url": api_url,
+        "limit": limit,
+        "pages": pages,
+        "raw_count": len(names),
+        "unique_count": len(set(names)),
+    }
+
+
+def doctor_check(command_text: str, timeout_seconds: int, required: bool) -> dict[str, Any]:
+    command_text = command_text.strip()
+    if not command_text:
+        return {"ok": True, "skipped": True, "reason": "empty GITEA_DOCTOR_CMD"}
+
+    command = shlex.split(command_text)
+    if not command:
+        return {"ok": True, "skipped": True, "reason": "empty parsed GITEA_DOCTOR_CMD"}
+
+    if not Path(command[0]).exists() and "/" in command[0]:
+        return {
+            "ok": not required,
+            "skipped": True,
+            "required": required,
+            "reason": f"doctor binary not found: {command[0]}",
+            "cmd": command,
+        }
+
+    result = run_command(command, timeout_seconds)
+    result["required"] = required
+    return result
+
+
+def compare_sets(left_name: str, left: set[str], right_name: str, right: set[str]) -> dict[str, Any]:
+    return {
+        f"{left_name}_missing_from_{right_name}": sorted(left - right),
+        f"{right_name}_missing_from_{left_name}": sorted(right - left),
+    }
+
+
+def build_report(args: argparse.Namespace) -> dict[str, Any]:
+    checks: dict[str, Any] = {}
+    errors: list[str] = []
+
+    git_names: set[str] = set()
+    db_names: set[str] = set()
+    api_names: set[str] = set()
+
+    try:
+        git_names, checks["git_refs"] = git_branch_names(args.repo_dir, args.timeout)
+    except Exception as exc:  # noqa: BLE001 - must preserve diagnostic text
+        errors.append(str(exc))
+        checks["git_refs"] = {"ok": False, "error": str(exc), "repo_dir": args.repo_dir}
+
+    try:
+        db_names, checks["sqlite_branch_metadata"] = sqlite_branch_names(args.db_path, args.repo_id)
+    except Exception as exc:  # noqa: BLE001
+        errors.append(str(exc))
+        checks["sqlite_branch_metadata"] = {"ok": False, "error": str(exc), "db_path": args.db_path}
+
+    if not args.skip_api:
+        try:
+            api_names, checks["gitea_branch_api"] = api_branch_names(
+                args.api_url,
+                args.api_limit,
+                args.timeout,
+                args.max_pages,
+            )
+        except Exception as exc:  # noqa: BLE001
+            errors.append(str(exc))
+            checks["gitea_branch_api"] = {"ok": False, "error": str(exc), "api_url": args.api_url}
+    else:
+        checks["gitea_branch_api"] = {"ok": True, "skipped": True}
+
+    if not args.skip_doctor:
+        checks["gitea_doctor"] = doctor_check(args.doctor_cmd, args.doctor_timeout, args.doctor_required)
+        if not checks["gitea_doctor"].get("ok", False):
+            errors.append("gitea doctor consistency check failed")
+    else:
+        checks["gitea_doctor"] = {"ok": True, "skipped": True}
+
+    drift: dict[str, Any] = {}
+    if checks.get("git_refs", {}).get("ok") and checks.get("sqlite_branch_metadata", {}).get("ok"):
+        drift.update(compare_sets("git", git_names, "db", db_names))
+    if checks.get("git_refs", {}).get("ok") and checks.get("gitea_branch_api", {}).get("ok") and not args.skip_api:
+        drift.update(compare_sets("git", git_names, "api", api_names))
+    if checks.get("sqlite_branch_metadata", {}).get("ok") and checks.get("gitea_branch_api", {}).get("ok") and not args.skip_api:
+        drift.update(compare_sets("db", db_names, "api", api_names))
+
+    drift_counts = {key: len(value) for key, value in drift.items()}
+    drift_found = any(drift_counts.values())
+
+    counts = {
+        "git_refs": len(git_names) if checks.get("git_refs", {}).get("ok") else None,
+        "db_live_branches": len(db_names) if checks.get("sqlite_branch_metadata", {}).get("ok") else None,
+        "api_visible_branches": len(api_names)
+        if checks.get("gitea_branch_api", {}).get("ok") and not args.skip_api
+        else None,
+    }
+
+    unhealthy_checks = [
+        name for name, result in checks.items()
+        if not result.get("ok", False)
+    ]
+
+    status = "unhealthy" if errors or drift_found or unhealthy_checks else "healthy"
+
+    return {
+        "service": "gitea-branch-consistency-sentinel",
+        "status": status,
+        "generated_at": utc_now(),
+        "repo": {
+            "repo_dir": args.repo_dir,
+            "db_path": args.db_path,
+            "repo_id": args.repo_id,
+            "api_url": args.api_url,
+        },
+        "counts": counts,
+        "checks": checks,
+        "drift": drift,
+        "drift_counts": drift_counts,
+        "errors": errors,
+        "unhealthy_checks": unhealthy_checks,
+    }
+
+
+def write_safe_mode_if_needed(report: dict[str, Any], args: argparse.Namespace) -> None:
+    safe_file = Path(args.safe_mode_file)
+    report_file = Path(args.report_file)
+
+    if report["status"] != "healthy":
+        if args.emit_safe_mode:
+            marker = {
+                "service": "autonomous-safe-mode",
+                "status": "active",
+                "entered_at": utc_now(),
+                "reason": "gitea_branch_consistency_unhealthy",
+                "source": "gitea-branch-consistency-sentinel",
+                "report_file": str(report_file),
+                "counts": report.get("counts", {}),
+                "drift_counts": report.get("drift_counts", {}),
+                "unhealthy_checks": report.get("unhealthy_checks", []),
+                "operator_action": (
+                    "Stop autonomous mutations, inspect Gitea refs/branch metadata/API drift, "
+                    "repair minimal inconsistent state, then run runtime acceptance tests."
+                ),
+            }
+            atomic_write_json(safe_file, marker)
+        return
+
+    if args.clear_safe_mode_if_healthy and safe_file.exists():
+        history_dir = safe_file.parent / "safe-mode-history"
+        history_dir.mkdir(parents=True, exist_ok=True)
+        archive = history_dir / f"{safe_file.stem}.cleared-{utc_now().replace(':', '').replace('-', '')}.json"
+        safe_file.replace(archive)
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Detect Gitea branch metadata drift and emit safe mode markers.")
+    parser.add_argument("--repo-dir", default=os.environ.get("GITEA_REPO_DIR", DEFAULT_REPO_DIR))
+    parser.add_argument("--db-path", default=os.environ.get("GITEA_DB_PATH", DEFAULT_DB_PATH))
+    parser.add_argument("--repo-id", type=int, default=int(os.environ.get("GITEA_REPO_ID", "1")))
+    parser.add_argument("--api-url", default=os.environ.get("GITEA_API_URL", DEFAULT_API_URL))
+    parser.add_argument("--api-limit", type=int, default=int(os.environ.get("GITEA_BRANCH_API_LIMIT", "50")))
+    parser.add_argument("--max-pages", type=int, default=int(os.environ.get("GITEA_BRANCH_API_MAX_PAGES", "200")))
+    parser.add_argument("--timeout", type=int, default=int(os.environ.get("GITEA_SENTINEL_TIMEOUT_SECONDS", "10")))
+    parser.add_argument("--doctor-timeout", type=int, default=int(os.environ.get("GITEA_DOCTOR_TIMEOUT_SECONDS", "45")))
+    parser.add_argument("--doctor-cmd", default=os.environ.get("GITEA_DOCTOR_CMD", DEFAULT_DOCTOR_CMD))
+    parser.add_argument("--doctor-required", action="store_true", default=env_bool("GITEA_DOCTOR_REQUIRED", False))
+    parser.add_argument("--skip-api", action="store_true", default=env_bool("GITEA_SENTINEL_SKIP_API", False))
+    parser.add_argument("--skip-doctor", action="store_true", default=env_bool("GITEA_SENTINEL_SKIP_DOCTOR", False))
+    parser.add_argument("--report-file", default=os.environ.get("GITEA_SENTINEL_REPORT_FILE", DEFAULT_REPORT_FILE))
+    parser.add_argument("--safe-mode-file", default=os.environ.get("AUTONOMOUS_SAFE_MODE_FILE", DEFAULT_SAFE_MODE_FILE))
+    parser.add_argument("--emit-safe-mode", action="store_true", help="Write safe-mode marker when unhealthy.")
+    parser.add_argument(
+        "--clear-safe-mode-if-healthy",
+        action="store_true",
+        help="Archive the existing safe-mode marker when all checks are healthy. Use only during controlled recovery.",
+    )
+    parser.add_argument("--json", action="store_true", help="Print full JSON report.")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str]) -> int:
+    args = parse_args(argv)
+    report = build_report(args)
+
+    report_path = Path(args.report_file)
+    atomic_write_json(report_path, report)
+    write_safe_mode_if_needed(report, args)
+
+    if args.json or report["status"] != "healthy":
+        json.dump(report, sys.stdout, ensure_ascii=False, indent=2, sort_keys=True)
+        sys.stdout.write("\n")
+    else:
+        print(
+            "healthy: "
+            f"git={report['counts']['git_refs']} "
+            f"db={report['counts']['db_live_branches']} "
+            f"api={report['counts']['api_visible_branches']}"
+        )
+
+    return 0 if report["status"] == "healthy" else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))