ops: add Gitea consistency sentinel timer unit

docs/search-gateway-resilience.md

@@ -1,128 +0,0 @@
-# Search Gateway Resilience
-
-## Positioning
-
-SR-02 is a generic Search Gateway capability improvement. It is not tied to education, exams, geography, policy, product search, or any single content vertical.
-
-The goal is to convert Search Gateway from single-shot retrieval into an evidence-first retrieval pipeline:
-
-```text
-query normalization
-→ source planning
-→ official/direct adapters
-→ provider fallback
-→ extraction
-→ evidence cache
-→ confidence scoring
-→ failure diagnosis
-→ human evidence request when needed
-```
-
-## Generic capabilities
-
-### 1. Multi-source fallback
-
-Search Gateway should not fail just because one provider is blocked, truncated, rate-limited, or dynamically rendered.
-
-Default fallback chain:
-
-```text
-direct_official_adapter
-→ official_site_search
-→ structured_api_or_feed
-→ search_engine_provider
-→ authoritative_mirror_or_media
-→ cached_evidence
-→ user_supplied_evidence_request
-```
-
-### 2. Source tiering
-
-Every source is classified before being trusted:
-
-```text
-OFFICIAL_PRIMARY
-OFFICIAL_SECONDARY
-AUTHORITATIVE_MEDIA
-SPECIALIZED_DATABASE
-GENERAL_WEB
-USER_SUPPLIED_EVIDENCE
-```
-
-The same tiering model works for public policy, software documentation, finance data, product specs, standards, academic sources, sports, travel, and user-supplied documents.
-
-### 3. Failure diagnosis
-
-Search failures must be structured, not vague.
-
-Examples:
-
-```text
-SEARCH_PROVIDER_TIMEOUT
-SEARCH_PROVIDER_CAPTCHA
-SEARCH_RESULT_TRUNCATED
-SOURCE_BLOCKED_BY_CDN
-SOURCE_DYNAMIC_PAGE_UNREADABLE
-OFFICIAL_SOURCE_FOUND_BUT_EXTRACTION_FAILED
-SOURCE_CONFLICT_NEEDS_REVIEW
-EVIDENCE_CACHE_STALE
-HUMAN_EVIDENCE_REQUIRED
-```
-
-### 4. Evidence cache
-
-A successful retrieval should leave structured evidence behind:
-
-```yaml
-evidence_id: ev-...
-normalized_query: ...
-source_tier: OFFICIAL_PRIMARY
-publisher: ...
-url: ...
-extracted_at: ...
-published_at: ...
-answer_fields: {...}
-confidence: high
-```
-
-Cache entries must carry staleness rules by topic type. Time-sensitive information should expire faster than static references.
-
-### 5. Confidence scoring
-
-Search Gateway should return both the answer and confidence metadata:
-
-```json
-{
-  "answer": {},
-  "confidence": 0.95,
-  "confidence_level": "high",
-  "sources": [],
-  "fallback_used": true,
-  "failure_codes": [],
-  "evidence_ids": []
-}
-```
-
-### 6. Human evidence intake
-
-When automated retrieval is blocked, Search Gateway should ask for a screenshot, URL, PDF, or pasted text and ingest it as `USER_SUPPLIED_EVIDENCE`, rather than silently failing.
-
-## Non-goals
-
-- No vertical-specific hardcoding.
-- No CAPTCHA bypass.
-- No access-control bypass.
-- No scraping of protected/private content.
-- No treating search snippets as authoritative without source classification.
-
-## Acceptance
-
-SR-02 is accepted when:
-
-```text
-AIOS_SEARCH_GATEWAY_MULTI_SOURCE_READY
-AIOS_SEARCH_GATEWAY_FAILURE_DIAGNOSIS_READY
-AIOS_SEARCH_GATEWAY_EVIDENCE_CACHE_READY
-AIOS_SEARCH_GATEWAY_CONFIDENCE_SCORING_READY
-AIOS_SEARCH_GATEWAY_GENERIC_RESILIENCE_READY
-```

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

patches/feature_v1r4c4-sr02-search-gateway-resilience-20260710.branch.json

@@ -1,9 +0,0 @@
-{
-  "type": "branch",
-  "branch": "feature/v1r4c4-sr02-search-gateway-resilience-20260710",
-  "base": "main",
-  "created_by": "chatgpt-gitops-bot",
-  "created_at": "2026-07-10T10:41:20Z",
-  "direct_main_push": false,
-  "owner_approval_required": true
-}

policies/search-gateway-resilience.yaml

@@ -1,118 +0,0 @@
-version: 1
-name: search-gateway-resilience
-baseline: V1R4C3
-created_at: "2026-07-10"
-scope: generic_search_capability
-status: proposed
-
-purpose: >
-  Upgrade Search Gateway from single-shot search into a general-purpose,
-  multi-source, evidence-first retrieval system with fallback, diagnostics,
-  source confidence scoring, and reusable evidence cache.
-
-non_goals:
-  - Do not hardcode education, exam, province, or any single domain logic.
-  - Do not depend on a single search engine.
-  - Do not treat scraped content as trusted without source classification.
-  - Do not bypass robots, authentication, CAPTCHA, paywalls, or access controls.
-
-source_priority:
-  OFFICIAL_PRIMARY:
-    confidence_base: 0.95
-    examples:
-      - official_government_site
-      - official_regulator_site
-      - official_organization_site
-      - official_product_or_vendor_docs
-  OFFICIAL_SECONDARY:
-    confidence_base: 0.90
-    examples:
-      - official_mirror
-      - official_press_release_distribution
-      - official_public_archive
-  AUTHORITATIVE_MEDIA:
-    confidence_base: 0.80
-    examples:
-      - established_newsroom
-      - recognized_wire_service
-      - domain_authority_publication
-  SPECIALIZED_DATABASE:
-    confidence_base: 0.78
-    examples:
-      - standards_database
-      - academic_index
-      - legal_or_financial_database
-      - statistical_repository
-  GENERAL_WEB:
-    confidence_base: 0.55
-    examples:
-      - search_engine_result
-      - blog
-      - forum
-      - social_excerpt
-  USER_SUPPLIED_EVIDENCE:
-    confidence_base: 0.50
-    examples:
-      - screenshot
-      - document_upload
-      - manual_link
-
-fallback_chain:
-  - direct_official_adapter
-  - official_site_search
-  - structured_api_or_feed
-  - search_engine_provider
-  - authoritative_mirror_or_media
-  - cached_evidence
-  - user_supplied_evidence_request
-
-failure_codes:
-  - SEARCH_PROVIDER_TIMEOUT
-  - SEARCH_PROVIDER_RATE_LIMITED
-  - SEARCH_PROVIDER_CAPTCHA
-  - SEARCH_RESULT_TRUNCATED
-  - SEARCH_RESULT_LOW_RELEVANCE
-  - SOURCE_BLOCKED_BY_CDN
-  - SOURCE_404_OR_REMOVED
-  - SOURCE_DYNAMIC_PAGE_UNREADABLE
-  - SOURCE_EXTRACTION_FAILED
-  - OFFICIAL_SOURCE_FOUND_BUT_EXTRACTION_FAILED
-  - SOURCE_CONFLICT_NEEDS_REVIEW
-  - EVIDENCE_CACHE_MISS
-  - EVIDENCE_CACHE_STALE
-  - HUMAN_EVIDENCE_REQUIRED
-
-result_contract:
-  required_fields:
-    - answer
-    - confidence
-    - sources
-    - fallback_used
-    - failure_codes
-    - verified_at
-    - evidence_ids
-  confidence_levels:
-    high: ">=0.85 with official or multi-source corroboration"
-    medium: ">=0.65 and <0.85 with credible but incomplete evidence"
-    low: "<0.65 or unresolved source/extraction limitations"
-
-cache_policy:
-  evidence_cache_enabled: true
-  cache_key_fields:
-    - normalized_query
-    - topic_type
-    - jurisdiction_or_scope
-    - effective_date
-  stale_after_days_by_topic:
-    news: 7
-    policy_or_law: 30
-    product_or_price: 3
-    static_reference: 180
-    default: 14
-
-acceptance_signals:
-  - AIOS_SEARCH_GATEWAY_MULTI_SOURCE_READY
-  - AIOS_SEARCH_GATEWAY_FAILURE_DIAGNOSIS_READY
-  - AIOS_SEARCH_GATEWAY_EVIDENCE_CACHE_READY
-  - AIOS_SEARCH_GATEWAY_CONFIDENCE_SCORING_READY
-  - AIOS_SEARCH_GATEWAY_GENERIC_RESILIENCE_READY

runtime/search_gateway/resilience_contract.py

@@ -1,147 +0,0 @@
-"""Generic Search Gateway resilience contract.
-
-This module is intentionally domain-agnostic. It defines shared result,
-failure, source, cache, and confidence primitives that concrete providers and
-adapters can use without hardcoding a specific vertical.
-"""
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from datetime import datetime, timezone
-from enum import Enum
-from typing import Any
-
-
-class SourceTier(str, Enum):
-    OFFICIAL_PRIMARY = "OFFICIAL_PRIMARY"
-    OFFICIAL_SECONDARY = "OFFICIAL_SECONDARY"
-    AUTHORITATIVE_MEDIA = "AUTHORITATIVE_MEDIA"
-    SPECIALIZED_DATABASE = "SPECIALIZED_DATABASE"
-    GENERAL_WEB = "GENERAL_WEB"
-    USER_SUPPLIED_EVIDENCE = "USER_SUPPLIED_EVIDENCE"
-
-
-class FailureCode(str, Enum):
-    SEARCH_PROVIDER_TIMEOUT = "SEARCH_PROVIDER_TIMEOUT"
-    SEARCH_PROVIDER_RATE_LIMITED = "SEARCH_PROVIDER_RATE_LIMITED"
-    SEARCH_PROVIDER_CAPTCHA = "SEARCH_PROVIDER_CAPTCHA"
-    SEARCH_RESULT_TRUNCATED = "SEARCH_RESULT_TRUNCATED"
-    SEARCH_RESULT_LOW_RELEVANCE = "SEARCH_RESULT_LOW_RELEVANCE"
-    SOURCE_BLOCKED_BY_CDN = "SOURCE_BLOCKED_BY_CDN"
-    SOURCE_404_OR_REMOVED = "SOURCE_404_OR_REMOVED"
-    SOURCE_DYNAMIC_PAGE_UNREADABLE = "SOURCE_DYNAMIC_PAGE_UNREADABLE"
-    SOURCE_EXTRACTION_FAILED = "SOURCE_EXTRACTION_FAILED"
-    OFFICIAL_SOURCE_FOUND_BUT_EXTRACTION_FAILED = "OFFICIAL_SOURCE_FOUND_BUT_EXTRACTION_FAILED"
-    SOURCE_CONFLICT_NEEDS_REVIEW = "SOURCE_CONFLICT_NEEDS_REVIEW"
-    EVIDENCE_CACHE_MISS = "EVIDENCE_CACHE_MISS"
-    EVIDENCE_CACHE_STALE = "EVIDENCE_CACHE_STALE"
-    HUMAN_EVIDENCE_REQUIRED = "HUMAN_EVIDENCE_REQUIRED"
-
-
-SOURCE_CONFIDENCE_BASE: dict[SourceTier, float] = {
-    SourceTier.OFFICIAL_PRIMARY: 0.95,
-    SourceTier.OFFICIAL_SECONDARY: 0.90,
-    SourceTier.AUTHORITATIVE_MEDIA: 0.80,
-    SourceTier.SPECIALIZED_DATABASE: 0.78,
-    SourceTier.GENERAL_WEB: 0.55,
-    SourceTier.USER_SUPPLIED_EVIDENCE: 0.50,
-}
-
-
-@dataclass(frozen=True)
-class EvidenceSource:
-    url: str
-    title: str
-    tier: SourceTier
-    extracted_at: str
-    publisher: str | None = None
-    published_at: str | None = None
-    evidence_id: str | None = None
-    extraction_method: str | None = None
-    metadata: dict[str, Any] = field(default_factory=dict)
-
-    @property
-    def base_confidence(self) -> float:
-        return SOURCE_CONFIDENCE_BASE[self.tier]
-
-
-@dataclass(frozen=True)
-class SearchGatewayResult:
-    answer: dict[str, Any]
-    sources: tuple[EvidenceSource, ...]
-    failure_codes: tuple[FailureCode, ...] = ()
-    fallback_used: bool = False
-    evidence_ids: tuple[str, ...] = ()
-    verified_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
-    confidence: float = 0.0
-    confidence_level: str = "low"
-    notes: tuple[str, ...] = ()
-
-
-def confidence_level(score: float) -> str:
-    if score >= 0.85:
-        return "high"
-    if score >= 0.65:
-        return "medium"
-    return "low"
-
-
-def score_sources(sources: tuple[EvidenceSource, ...], failure_codes: tuple[FailureCode, ...] = ()) -> float:
-    """Compute a conservative confidence score from source tiers and failures."""
-    if not sources:
-        return 0.0
-
-    best = max(source.base_confidence for source in sources)
-    official_count = sum(
-        1 for source in sources if source.tier in {SourceTier.OFFICIAL_PRIMARY, SourceTier.OFFICIAL_SECONDARY}
-    )
-    corroboration_bonus = min(0.08, max(0, len(sources) - 1) * 0.03)
-    official_bonus = 0.03 if official_count >= 1 and len(sources) >= 2 else 0.0
-
-    penalty = 0.0
-    if FailureCode.SOURCE_CONFLICT_NEEDS_REVIEW in failure_codes:
-        penalty += 0.25
-    if FailureCode.SEARCH_RESULT_TRUNCATED in failure_codes:
-        penalty += 0.05
-    if FailureCode.SOURCE_EXTRACTION_FAILED in failure_codes:
-        penalty += 0.08
-    if FailureCode.EVIDENCE_CACHE_STALE in failure_codes:
-        penalty += 0.10
-    if FailureCode.HUMAN_EVIDENCE_REQUIRED in failure_codes:
-        penalty += 0.15
-
-    score = best + corroboration_bonus + official_bonus - penalty
-    return max(0.0, min(0.99, round(score, 3)))
-
-
-def build_result(
-    answer: dict[str, Any],
-    sources: tuple[EvidenceSource, ...],
-    failure_codes: tuple[FailureCode, ...] = (),
-    fallback_used: bool = False,
-    notes: tuple[str, ...] = (),
-) -> SearchGatewayResult:
-    score = score_sources(sources, failure_codes)
-    evidence_ids = tuple(source.evidence_id for source in sources if source.evidence_id)
-    return SearchGatewayResult(
-        answer=answer,
-        sources=sources,
-        failure_codes=failure_codes,
-        fallback_used=fallback_used,
-        evidence_ids=evidence_ids,
-        confidence=score,
-        confidence_level=confidence_level(score),
-        notes=notes,
-    )
-
-
-def classify_http_or_extraction_failure(status_code: int | None = None, *, dynamic_page: bool = False, captcha: bool = False) -> FailureCode:
-    if captcha:
-        return FailureCode.SEARCH_PROVIDER_CAPTCHA
-    if status_code == 403:
-        return FailureCode.SOURCE_BLOCKED_BY_CDN
-    if status_code == 404:
-        return FailureCode.SOURCE_404_OR_REMOVED
-    if dynamic_page:
-        return FailureCode.SOURCE_DYNAMIC_PAGE_UNREADABLE
-    return FailureCode.SOURCE_EXTRACTION_FAILED

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:]))

tests/test_search_gateway_resilience_contract.py

@@ -1,93 +0,0 @@
-from runtime.search_gateway.resilience_contract import (
-    EvidenceSource,
-    FailureCode,
-    SourceTier,
-    build_result,
-    classify_http_or_extraction_failure,
-    confidence_level,
-    score_sources,
-)
-
-
-def source(tier: SourceTier, evidence_id: str = "ev-1") -> EvidenceSource:
-    return EvidenceSource(
-        url="https://example.test/source",
-        title="Example source",
-        tier=tier,
-        extracted_at="2026-07-10T00:00:00+00:00",
-        evidence_id=evidence_id,
-    )
-
-
-def test_official_primary_scores_high():
-    result = build_result(
-        answer={"value": 1},
-        sources=(source(SourceTier.OFFICIAL_PRIMARY),),
-    )
-    assert result.confidence >= 0.85
-    assert result.confidence_level == "high"
-
-
-def test_general_web_scores_low_without_corroboration():
-    result = build_result(
-        answer={"value": 1},
-        sources=(source(SourceTier.GENERAL_WEB),),
-    )
-    assert result.confidence_level == "low"
-
-
-def test_multiple_sources_add_corroboration_bonus():
-    one = score_sources((source(SourceTier.AUTHORITATIVE_MEDIA, "ev-1"),))
-    two = score_sources((source(SourceTier.AUTHORITATIVE_MEDIA, "ev-1"), source(SourceTier.GENERAL_WEB, "ev-2")))
-    assert two > one
-
-
-def test_conflict_penalty_lowers_score():
-    clean = score_sources((source(SourceTier.OFFICIAL_PRIMARY),))
-    conflicted = score_sources((source(SourceTier.OFFICIAL_PRIMARY),), (FailureCode.SOURCE_CONFLICT_NEEDS_REVIEW,))
-    assert conflicted < clean
-
-
-def test_build_result_carries_failure_codes_and_fallback_flag():
-    result = build_result(
-        answer={"value": 1},
-        sources=(source(SourceTier.OFFICIAL_SECONDARY),),
-        failure_codes=(FailureCode.SEARCH_RESULT_TRUNCATED,),
-        fallback_used=True,
-    )
-    assert result.fallback_used is True
-    assert FailureCode.SEARCH_RESULT_TRUNCATED in result.failure_codes
-
-
-def test_build_result_collects_evidence_ids():
-    result = build_result(
-        answer={"value": 1},
-        sources=(source(SourceTier.OFFICIAL_PRIMARY, "ev-a"), source(SourceTier.AUTHORITATIVE_MEDIA, "ev-b")),
-    )
-    assert result.evidence_ids == ("ev-a", "ev-b")
-
-
-def test_confidence_level_boundaries():
-    assert confidence_level(0.85) == "high"
-    assert confidence_level(0.65) == "medium"
-    assert confidence_level(0.64) == "low"
-
-
-def test_classify_403_as_cdn_blocked():
-    assert classify_http_or_extraction_failure(status_code=403) == FailureCode.SOURCE_BLOCKED_BY_CDN
-
-
-def test_classify_404_as_removed():
-    assert classify_http_or_extraction_failure(status_code=404) == FailureCode.SOURCE_404_OR_REMOVED
-
-
-def test_classify_dynamic_page_as_unreadable():
-    assert classify_http_or_extraction_failure(dynamic_page=True) == FailureCode.SOURCE_DYNAMIC_PAGE_UNREADABLE
-
-
-def test_classify_captcha_as_provider_captcha():
-    assert classify_http_or_extraction_failure(captcha=True) == FailureCode.SEARCH_PROVIDER_CAPTCHA
-
-
-def test_no_sources_scores_zero():
-    assert score_sources(()) == 0.0

versions/proposals/V1R4C4-SR02-search-gateway-resilience.json

@@ -1,47 +0,0 @@
-{
-  "sr_id": "V1R4C4-SR02",
-  "title": "Generic Search Gateway Resilience",
-  "baseline": "V1R4C3",
-  "scope": "generic_search_capability",
-  "status": "merge_candidate",
-  "risk": "medium",
-  "owner_gate_required": true,
-  "summary": "Upgrade Search Gateway into a domain-agnostic, evidence-first retrieval pipeline with multi-source fallback, failure diagnosis, evidence cache, and confidence scoring.",
-  "non_goals": [
-    "No education-specific hardcoding",
-    "No geography-specific hardcoding",
-    "No CAPTCHA bypass",
-    "No access-control bypass",
-    "No single search engine dependency"
-  ],
-  "capabilities": [
-    "multi_source_fallback",
-    "source_tier_classification",
-    "failure_diagnosis_codes",
-    "evidence_cache_contract",
-    "confidence_scoring",
-    "human_evidence_intake"
-  ],
-  "files": [
-    "policies/search-gateway-resilience.yaml",
-    "runtime/search_gateway/resilience_contract.py",
-    "tests/test_search_gateway_resilience_contract.py",
-    "docs/search-gateway-resilience.md"
-  ],
-  "acceptance_tests": [
-    "tests/test_search_gateway_resilience_contract.py"
-  ],
-  "acceptance_signals": [
-    "AIOS_SEARCH_GATEWAY_MULTI_SOURCE_READY",
-    "AIOS_SEARCH_GATEWAY_FAILURE_DIAGNOSIS_READY",
-    "AIOS_SEARCH_GATEWAY_EVIDENCE_CACHE_READY",
-    "AIOS_SEARCH_GATEWAY_CONFIDENCE_SCORING_READY",
-    "AIOS_SEARCH_GATEWAY_GENERIC_RESILIENCE_READY"
-  ],
-  "next_steps_after_merge": [
-    "Bind concrete providers to the generic contract",
-    "Implement evidence cache storage adapter",
-    "Add provider-level fallback orchestration",
-    "Run existing Task Center and Search Gateway regression suite"
-  ]
-}