deploy: fefbb688 feat(step3): Gnosis sink + bottom-lane S3 + Brave grounding (#190)

elpidaapp/chat_engine.py

@@ -263,13 +263,27 @@ def needs_grounding(text: str) -> bool:
     return any(kw in text_lower for kw in _GROUNDING_KEYWORDS)
 
 
+def _strip_grounding_provider_footer(result: str) -> Tuple[str, str]:
+    """Remove the authoritative provider footer from ground_query output."""
+    provider = "duckduckgo"
+    cleaned_result = result.rstrip()
+    last_newline = cleaned_result.rfind("\n")
+    footer = cleaned_result[last_newline + 1:].strip()
+    prefix = "<!-- provider: "
+    suffix = " -->"
+    if footer.startswith(prefix) and footer.endswith(suffix):
+        provider = footer[len(prefix):-len(suffix)].strip() or provider
+        cleaned_result = "" if last_newline == -1 else cleaned_result[:last_newline]
+    return cleaned_result, provider
+
+
 def fetch_live_context(
     query: str,
     llm,
     prefer_grok: bool = False,
 ) -> Tuple[str, str]:
     """
-    Fetch live grounding — DuckDuckGo/Wikipedia first (free), then
+    Fetch live grounding — Brave/DuckDuckGo/Wikipedia first (free), then
     Perplexity/Grok LLM fallback (paid) only if free search returns empty.
     Returns (context_text, provider_used).
     One call per turn max — budget aware.
@@ -277,11 +291,12 @@ def fetch_live_context(
     # ── Primary: free web search via domain_grounding ──────────────
     try:
         from elpidaapp.domain_grounding import ground_query
-        ddg_result = ground_query(query, max_results=3)
-        if ddg_result and len(ddg_result.strip()) > 30:
-            return ddg_result.strip(), "duckduckgo"
+        grounding_result = ground_query(query, max_results=3)
+        if grounding_result and len(grounding_result.strip()) > 30:
+            cleaned_result, provider = _strip_grounding_provider_footer(grounding_result)
+            return cleaned_result.strip(), provider
     except Exception as e:
-        logger.debug("DDG grounding failed, falling back to LLM: %s", e)
+        logger.debug("Free grounding failed, falling back to LLM: %s", e)
 
     # ── Fallback: LLM-based grounding (Perplexity/Grok) ───────────
     grounding_prompt = (

elpidaapp/contradiction_log.py

@@ -33,10 +33,11 @@ Location: cache/contradiction_log.jsonl
 
 import json
 import logging
+import os
 from collections import deque
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Iterable, List, Optional
 
 logger = logging.getLogger("elpida.contradiction_log")
 
@@ -45,6 +46,62 @@ CONTRADICTION_LOG = CACHE_DIR / "contradiction_log.jsonl"
 
 # In-memory sliding window for recent query
 MAX_IN_MEMORY = 200
+S3_BUCKET = os.environ.get("CONTRADICTION_S3_BUCKET", "elpida-body-evolution")
+S3_REGION = os.environ.get("CONTRADICTION_S3_REGION", "eu-north-1")
+S3_KEY = os.environ.get("CONTRADICTION_S3_KEY", "federation/contradiction_log.jsonl")
+_MISSING_S3_KEY_CODES = {"NoSuchKey", "404", "NotFound"}
+
+
+def _is_missing_s3_key_error(s3, exc: Exception) -> bool:
+    no_such_key = getattr(getattr(s3, "exceptions", None), "NoSuchKey", None)
+    if isinstance(no_such_key, type) and isinstance(exc, no_such_key):
+        return True
+    response = getattr(exc, "response", None)
+    if isinstance(response, dict):
+        code = response.get("Error", {}).get("Code")
+        return str(code) in _MISSING_S3_KEY_CODES
+    return False
+
+
+def _has_s3_credentials() -> bool:
+    return bool(
+        os.environ.get("AWS_ACCESS_KEY_ID")
+        and os.environ.get("AWS_SECRET_ACCESS_KEY")
+    )
+
+
+def _merge_jsonl_payloads(*payloads: bytes) -> bytes:
+    """Merge JSONL bodies without dropping rows present only in local cache."""
+    merged_lines: List[str] = []
+    seen = set()
+    for payload in payloads:
+        text = payload.decode("utf-8", errors="replace")
+        for line in text.splitlines():
+            if not line.strip() or line in seen:
+                continue
+            merged_lines.append(line)
+            seen.add(line)
+    return ("\n".join(merged_lines) + ("\n" if merged_lines else "")).encode("utf-8")
+
+
+def _drop_trailing_jsonl_lines(payload: bytes, lines_to_drop: Iterable[str]) -> bytes:
+    drop_counts: Dict[str, int] = {}
+    for line in lines_to_drop:
+        key = line.rstrip("\n")
+        drop_counts[key] = drop_counts.get(key, 0) + 1
+    if not drop_counts:
+        return payload
+
+    text = payload.decode("utf-8", errors="replace")
+    kept_reversed: List[str] = []
+    for line in reversed(text.splitlines()):
+        count = drop_counts.get(line, 0)
+        if count:
+            drop_counts[line] = count - 1
+            continue
+        kept_reversed.append(line)
+    kept = list(reversed(kept_reversed))
+    return ("\n".join(kept) + ("\n" if kept else "")).encode("utf-8")
 
 
 class ContradictionLog:
@@ -66,6 +123,49 @@ class ContradictionLog:
         self._entries: deque = deque(maxlen=MAX_IN_MEMORY)
         self._total: int = 0
         self._type_counts: Dict[str, int] = {}
+        self._s3 = None
+        self._pending_lines: List[str] = []
+        self._hydrate_from_disk()
+
+    def _hydrate_from_disk(self) -> None:
+        """Initialize counters from the restored append-only ledger."""
+        if not CONTRADICTION_LOG.exists():
+            return
+        records = 0
+        max_number = 0
+        type_counts: Dict[str, int] = {}
+        try:
+            with open(CONTRADICTION_LOG, "r", encoding="utf-8") as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    try:
+                        entry = json.loads(line)
+                    except json.JSONDecodeError:
+                        continue
+                    records += 1
+                    self._entries.append(entry)
+                    try:
+                        max_number = max(
+                            max_number,
+                            int(entry.get("contradiction_number", 0)),
+                        )
+                    except (TypeError, ValueError):
+                        pass
+                    contradiction_type = entry.get("type")
+                    if contradiction_type:
+                        type_counts[contradiction_type] = (
+                            type_counts.get(contradiction_type, 0) + 1
+                        )
+            self._total = max(records, max_number)
+            self._type_counts = type_counts
+            logger.info(
+                "ContradictionLog: loaded %d existing records from %s",
+                records, CONTRADICTION_LOG,
+            )
+        except Exception as e:
+            logger.warning("ContradictionLog: could not read ledger: %s", e)
 
     def record(
         self,
@@ -111,6 +211,7 @@ class ContradictionLog:
                 f.write(json.dumps(entry, ensure_ascii=False) + "\n")
         except Exception as e:
             logger.debug("Contradiction persist failed: %s", e)
+        self._queue_for_s3(entry)
 
         return entry
 
@@ -180,3 +281,78 @@ class ContradictionLog:
             "type_counts": dict(self._type_counts),
             "recent_count": len(self._entries),
         }
+
+    def _get_s3(self):
+        if self._s3 is not None:
+            return self._s3
+        sink_enabled = os.environ.get("CONTRADICTION_S3_SINK", "1").strip().lower()
+        if sink_enabled in {"0", "false", "no", "off", ""}:
+            return None
+        if not _has_s3_credentials():
+            return None
+        try:
+            import boto3
+            self._s3 = boto3.client("s3", region_name=S3_REGION)
+            return self._s3
+        except Exception as e:
+            logger.debug("Contradiction S3 client unavailable: %s", e)
+            return None
+
+    def _queue_for_s3(self, entry: Dict[str, Any]) -> None:
+        sink_enabled = os.environ.get("CONTRADICTION_S3_SINK", "1").strip().lower()
+        if sink_enabled in {"0", "false", "no", "off", ""}:
+            return
+        if getattr(self, "_s3", None) is None and not _has_s3_credentials():
+            return
+        if not hasattr(self, "_pending_lines"):
+            self._pending_lines = []
+        line = json.dumps(entry, ensure_ascii=False) + "\n"
+        self._pending_lines.append(line)
+
+    def flush_pending_to_s3(self, max_lines: int = 200) -> None:
+        """Flush queued contradiction records to S3 on heartbeat cadence."""
+        pending = getattr(self, "_pending_lines", [])
+        if not pending:
+            return
+        s3 = self._get_s3()
+        if not s3:
+            return
+        batch_size = max(1, max_lines)
+        batch = pending[:batch_size]
+        try:
+            try:
+                resp = s3.get_object(Bucket=S3_BUCKET, Key=S3_KEY)
+                existing = resp["Body"].read()
+            except Exception as e:
+                if _is_missing_s3_key_error(s3, e):
+                    existing = b""
+                else:
+                    raise
+            try:
+                local_payload = CONTRADICTION_LOG.read_bytes()
+            except OSError:
+                local_payload = b""
+            local_payload = _drop_trailing_jsonl_lines(
+                local_payload,
+                pending[len(batch):],
+            )
+            pending_payload = "".join(batch).encode("utf-8")
+            body = _merge_jsonl_payloads(
+                existing,
+                local_payload,
+                pending_payload,
+            )
+            s3.put_object(
+                Bucket=S3_BUCKET,
+                Key=S3_KEY,
+                Body=body,
+                ContentType="application/x-ndjson",
+            )
+            del pending[: len(batch)]
+        except Exception as e:
+            logger.debug("Contradiction S3 flush failed: %s", e)
+
+    def _append_to_s3(self, entry: Dict[str, Any]) -> None:
+        """Compatibility path; runtime uses heartbeat flush."""
+        self._queue_for_s3(entry)
+        self.flush_pending_to_s3(max_lines=1)

elpidaapp/divergence_engine.py

@@ -20,9 +20,11 @@ import sys
 import os
 import json
 import time
+import hashlib
 import logging
 import concurrent.futures
-from datetime import datetime
+import uuid
+from datetime import datetime, timezone
 from typing import Dict, List, Optional, Any
 from pathlib import Path
 
@@ -34,6 +36,63 @@ from elpida_config import DOMAINS, AXIOMS, AXIOM_RATIOS
 
 logger = logging.getLogger("elpidaapp.divergence")
 
+
+def _push_audit_record_to_s3(record: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    """Best-effort WORLD audit sink. Never raises on failure."""
+    sink_enabled = os.environ.get("DIVERGENCE_AUDIT_S3_SINK", "1").strip().lower()
+    if sink_enabled in {"0", "false", "no", "off", ""}:
+        return None
+
+    if not (
+        os.environ.get("AWS_ACCESS_KEY_ID")
+        and os.environ.get("AWS_SECRET_ACCESS_KEY")
+    ):
+        return None
+
+    try:
+        import boto3
+        from botocore.exceptions import BotoCoreError, ClientError
+    except Exception:
+        return None
+
+    bucket = os.environ.get("DIVERGENCE_AUDIT_BUCKET", "elpida-external-interfaces")
+    region = os.environ.get("DIVERGENCE_AUDIT_REGION", "eu-north-1")
+    prefix = os.environ.get("DIVERGENCE_AUDIT_PREFIX", "audits/")
+    if not prefix.endswith("/"):
+        prefix += "/"
+
+    try:
+        body = json.dumps(record, ensure_ascii=False).encode("utf-8")
+        ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+        problem_hash = hashlib.sha256(
+            str(record.get("problem", "")).encode("utf-8")
+        ).hexdigest()[:12]
+        nonce = uuid.uuid4().hex
+        key = f"{prefix}{ts}_{problem_hash}_{nonce}.json"
+        s3 = boto3.client("s3", region_name=region)
+        resp = s3.put_object(
+            Bucket=bucket,
+            Key=key,
+            Body=body,
+            ContentType="application/json; charset=utf-8",
+            Metadata={
+                "elpida-record-type": "audit",
+                "elpida-falsifiable-core": "true",
+            },
+        )
+        return {
+            "bucket": bucket,
+            "key": key,
+            "etag": (resp.get("ETag") or "").strip('"'),
+            "size_bytes": len(body),
+        }
+    except (BotoCoreError, ClientError) as e:
+        logger.warning("Audit S3 sink failed: %s", e)
+        return None
+    except Exception as e:  # pragma: no cover - defensive never-raise
+        logger.warning("Audit S3 sink unexpected error: %s", e)
+        return None
+
 # ── Integration layer (lazy imports — optional dependencies) ──
 _governance_client = None
 _frozen_mind = None
@@ -364,6 +423,27 @@ class DivergenceEngine:
             "kaya_events": kaya_events,
         }
 
+        result["falsifiable_core"] = {
+            "single_model": baseline,
+            "domain_responses": domain_responses,
+            "divergence": divergence,
+        }
+        result["unfalsifiable_frame"] = {
+            "falsifiable": False,
+            "synthesis": synthesis,
+        }
+        audit_record = {
+            "record_type": "divergence_audit",
+            "problem": problem,
+            "timestamp": ts,
+            "total_time_s": elapsed,
+            "falsifiable_core": result["falsifiable_core"],
+            "unfalsifiable_frame": result["unfalsifiable_frame"],
+            "governance_check": governance_check,
+            "frozen_mind_context": frozen_mind_context,
+        }
+        result["audit_s3_receipt"] = _push_audit_record_to_s3(audit_record)
+
         if save_to:
             path = Path(save_to)
             path.parent.mkdir(parents=True, exist_ok=True)

elpidaapp/domain_grounding.py

@@ -3,10 +3,11 @@ Domain Internet Grounding
 =========================
 
 Gives Elpida's 16 domains access to live web data.
-Two search backends, automatic failover:
+Three search backends, automatic failover:
 
-  1. DuckDuckGo text search (primary — zero API keys)
-  2. Wikipedia API (fallback — always available, English content)
+  1. Brave Search API (primary when BRAVE_API_KEY is configured)
+  2. DuckDuckGo text search (zero-key fallback)
+  3. Wikipedia API (fallback — always available, English content)
 
 Each domain query can be optionally augmented with real-world context
 before the LLM prompt is built. The grounding fetches 3-5 results
@@ -27,11 +28,12 @@ Architecture:
 """
 
 import logging
+import os
 import re
 import time
 import threading
-from functools import lru_cache
-from typing import Optional, List, Dict
+from collections import OrderedDict
+from typing import Optional, List, Dict, Tuple
 
 import requests
 
@@ -42,6 +44,52 @@ _lock = threading.Lock()
 _last_search_time = 0.0
 _RATE_LIMIT_S = 3.0
 _TIMEOUT_S = 8
+_GROUNDING_COUNTS = {
+    "brave_hits": 0,
+    "ddg_hits": 0,
+    "wiki_fallbacks": 0,
+    "empty_results": 0,
+}
+_GROUNDING_COUNTS_LOCK = threading.Lock()
+_GROUNDING_CACHE: "OrderedDict[Tuple[str, int, bool], str]" = OrderedDict()
+_GROUNDING_CACHE_MAXSIZE = 128
+_GROUNDING_CACHE_LOCK = threading.Lock()
+
+
+def _bump_counter(counter: str) -> None:
+    with _GROUNDING_COUNTS_LOCK:
+        _GROUNDING_COUNTS[counter] = _GROUNDING_COUNTS.get(counter, 0) + 1
+
+
+def grounding_stats() -> Dict[str, int]:
+    """Return backend hit/fallback counters for runtime visibility."""
+    with _GROUNDING_COUNTS_LOCK:
+        return dict(_GROUNDING_COUNTS)
+
+
+def _brave_configured() -> bool:
+    return bool(os.environ.get("BRAVE_API_KEY", "").strip())
+
+
+def _cache_get(key: Tuple[str, int, bool]) -> Optional[str]:
+    with _GROUNDING_CACHE_LOCK:
+        cached = _GROUNDING_CACHE.get(key)
+        if cached is not None:
+            _GROUNDING_CACHE.move_to_end(key)
+        return cached
+
+
+def _cache_set(key: Tuple[str, int, bool], value: str) -> None:
+    with _GROUNDING_CACHE_LOCK:
+        _GROUNDING_CACHE[key] = value
+        _GROUNDING_CACHE.move_to_end(key)
+        while len(_GROUNDING_CACHE) > _GROUNDING_CACHE_MAXSIZE:
+            _GROUNDING_CACHE.popitem(last=False)
+
+
+def _cache_clear() -> None:
+    with _GROUNDING_CACHE_LOCK:
+        _GROUNDING_CACHE.clear()
 
 
 def _rate_limit():
@@ -75,7 +123,11 @@ def _search_ddg(query: str, max_results: int) -> List[Dict[str, str]]:
             if _is_english(r.get("title", "") + r.get("body", ""))
         ]
         return [
-            {"title": r.get("title", ""), "body": r.get("body", "")}
+            {
+                "title": r.get("title", ""),
+                "body": r.get("body", ""),
+                "href": r.get("href", ""),
+            }
             for r in english[:max_results]
         ]
     except Exception as e:
@@ -83,6 +135,42 @@ def _search_ddg(query: str, max_results: int) -> List[Dict[str, str]]:
         return []
 
 
+def _search_brave(query: str, max_results: int) -> List[Dict[str, str]]:
+    """Search via Brave Search API when BRAVE_API_KEY is configured."""
+    api_key = os.environ.get("BRAVE_API_KEY", "").strip()
+    if not api_key:
+        return []
+
+    try:
+        resp = requests.get(
+            "https://api.search.brave.com/res/v1/web/search",
+            params={"q": query, "count": max_results},
+            headers={
+                "Accept": "application/json",
+                "X-Subscription-Token": api_key,
+            },
+            timeout=_TIMEOUT_S,
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        results = []
+        for item in data.get("web", {}).get("results", []):
+            title = item.get("title", "")
+            description = item.get("description", "")
+            if title and description and _is_english(title + description):
+                results.append(
+                    {
+                        "title": title,
+                        "body": description,
+                        "href": item.get("url", ""),
+                    }
+                )
+        return results[:max_results]
+    except Exception as e:
+        logger.debug("Brave search failed: %s", e)
+        return []
+
+
 def _search_wikipedia(query: str, max_results: int) -> List[Dict[str, str]]:
     """Search via Wikipedia API (always English, always available)."""
     try:
@@ -107,19 +195,24 @@ def _search_wikipedia(query: str, max_results: int) -> List[Dict[str, str]]:
             # Strip HTML tags from snippet
             snippet = re.sub(r'<[^>]+>', '', item.get("snippet", ""))
             if title and snippet:
-                results.append({"title": title, "body": snippet})
+                results.append(
+                    {
+                        "title": title,
+                        "body": snippet,
+                        "href": f"https://en.wikipedia.org/wiki/{title.replace(' ', '_')}",
+                    }
+                )
         return results[:max_results]
     except Exception as e:
         logger.debug("Wikipedia search failed: %s", e)
         return []
 
 
-@lru_cache(maxsize=128)
 def ground_query(query: str, max_results: int = 3) -> str:
     """
     Search the web for context relevant to a domain query.
 
-    Tries DuckDuckGo first, falls back to Wikipedia API.
+    Tries Brave first when configured, then DuckDuckGo, then Wikipedia.
 
     Args:
         query: The search query (typically the problem + domain keywords)
@@ -128,16 +221,39 @@ def ground_query(query: str, max_results: int = 3) -> str:
     Returns:
         Formatted string of web snippets, or "" on any failure.
     """
-    _rate_limit()
+    brave_enabled = _brave_configured()
+    cache_key = (query, max_results, brave_enabled)
+    cached = _cache_get(cache_key)
+    if cached is not None:
+        return cached
 
-    # Try DuckDuckGo first
-    results = _search_ddg(query, max_results)
+    _rate_limit()
 
-    # Fallback to Wikipedia if DDG returned nothing
+    provider = "unknown"
+    cacheable = False
+    # Try Brave first when key is available.
+    results = _search_brave(query, max_results)
+    if results:
+        _bump_counter("brave_hits")
+        provider = "brave"
+        cacheable = True
+    else:
+        # Then DuckDuckGo.
+        results = _search_ddg(query, max_results)
+        if results:
+            _bump_counter("ddg_hits")
+            provider = "duckduckgo"
+            cacheable = not brave_enabled
+
+    # Fallback to Wikipedia if Brave/DDG returned nothing.
     if not results:
         results = _search_wikipedia(query, max_results)
+        if results:
+            _bump_counter("wiki_fallbacks")
+            provider = "wikipedia"
 
     if not results:
+        _bump_counter("empty_results")
         return ""
 
     snippets = []
@@ -150,11 +266,17 @@ def ground_query(query: str, max_results: int = 3) -> str:
     if not snippets:
         return ""
 
-    return (
+    output = (
         "─── LIVE WEB CONTEXT ───\n"
         + "\n".join(snippets)
-        + "\n─── END WEB CONTEXT ───"
+        + f"\n─── END WEB CONTEXT ───\n<!-- provider: {provider} -->"
     )
+    if cacheable:
+        _cache_set(cache_key, output)
+    return output
+
+
+ground_query.cache_clear = _cache_clear  # type: ignore[attr-defined]
 
 
 def ground_domain_query(

elpidaapp/governance_event_spine.py

@@ -10,16 +10,94 @@ from __future__ import annotations
 
 import hashlib
 import json
+import logging
+import os
 from dataclasses import asdict, dataclass
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Dict, Iterable, List, Optional
 
+logger = logging.getLogger("elpida.governance_event_spine")
+
+S3_BUCKET = os.environ.get("SPINE_S3_BUCKET", "elpida-body-evolution")
+S3_REGION = os.environ.get("SPINE_S3_REGION", "eu-north-1")
+S3_KEY_MAP = {
+    "internal_dilemmas": "federation/internal_dilemmas.jsonl",
+    "oracle_advisories": "federation/oracle_advisories.jsonl",
+    "synthesis_council_decisions": "federation/synthesis_council_decisions.jsonl",
+    "synthesis_resolutions": "federation/synthesis_resolutions.jsonl",
+    "federation_events": "federation/federation_events.jsonl",
+    "ark_updates": "federation/ark_updates.jsonl",
+    "federation_stats": "federation/federation_stats.jsonl",
+}
+S3_PATH_ATTR_MAP = {
+    "internal_dilemmas": "internal_dilemmas_path",
+    "oracle_advisories": "oracle_advisories_path",
+    "synthesis_council_decisions": "synthesis_council_decisions_path",
+    "synthesis_resolutions": "synthesis_resolutions_path",
+    "federation_events": "federation_events_path",
+    "ark_updates": "ark_updates_path",
+    "federation_stats": "federation_stats_path",
+}
+_MISSING_S3_KEY_CODES = {"NoSuchKey", "404", "NotFound"}
+
 
 def _utc_now() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
+def _is_missing_s3_key_error(s3, exc: Exception) -> bool:
+    no_such_key = getattr(getattr(s3, "exceptions", None), "NoSuchKey", None)
+    if isinstance(no_such_key, type) and isinstance(exc, no_such_key):
+        return True
+    response = getattr(exc, "response", None)
+    if isinstance(response, dict):
+        code = response.get("Error", {}).get("Code")
+        return str(code) in _MISSING_S3_KEY_CODES
+    return False
+
+
+def _has_s3_credentials() -> bool:
+    return bool(
+        os.environ.get("AWS_ACCESS_KEY_ID")
+        and os.environ.get("AWS_SECRET_ACCESS_KEY")
+    )
+
+
+def _merge_jsonl_payloads(*payloads: bytes) -> bytes:
+    """Merge JSONL bodies without dropping rows present only in local cache."""
+    merged_lines: List[str] = []
+    seen = set()
+    for payload in payloads:
+        text = payload.decode("utf-8", errors="replace")
+        for line in text.splitlines():
+            if not line.strip() or line in seen:
+                continue
+            merged_lines.append(line)
+            seen.add(line)
+    return ("\n".join(merged_lines) + ("\n" if merged_lines else "")).encode("utf-8")
+
+
+def _drop_trailing_jsonl_lines(payload: bytes, lines_to_drop: Iterable[str]) -> bytes:
+    drop_counts: Dict[str, int] = {}
+    for line in lines_to_drop:
+        key = line.rstrip("\n")
+        drop_counts[key] = drop_counts.get(key, 0) + 1
+    if not drop_counts:
+        return payload
+
+    text = payload.decode("utf-8", errors="replace")
+    kept_reversed: List[str] = []
+    for line in reversed(text.splitlines()):
+        count = drop_counts.get(line, 0)
+        if count:
+            drop_counts[line] = count - 1
+            continue
+        kept_reversed.append(line)
+    kept = list(reversed(kept_reversed))
+    return ("\n".join(kept) + ("\n" if kept else "")).encode("utf-8")
+
+
 def _stable_id(prefix: str, *parts: Any) -> str:
     raw = "::".join(str(part) for part in parts if part is not None)
     digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:12]
@@ -103,6 +181,97 @@ class GovernanceEventSpine:
         self.ark_updates_path = self.base_dir / "ark_updates.jsonl"
         self.federation_stats_path = self.base_dir / "federation_stats.jsonl"
         self.living_axioms_path = self.base_dir / "living_axioms.jsonl"
+        self._s3 = None
+        self._pending_lines: Dict[str, List[str]] = {}
+
+    def _append_stream(self, path: Path, payload: Dict[str, Any], stream: Optional[str] = None) -> None:
+        _append_jsonl(path, payload)
+        if stream:
+            self._queue_for_s3(stream, payload)
+
+    def _queue_for_s3(self, stream: str, payload: Dict[str, Any]) -> None:
+        if stream not in S3_KEY_MAP:
+            return
+        sink_enabled = os.environ.get("SPINE_S3_SINK", "1").strip().lower()
+        if sink_enabled in {"0", "false", "no", "off", ""}:
+            return
+        if getattr(self, "_s3", None) is None and not _has_s3_credentials():
+            return
+        line = json.dumps(payload, ensure_ascii=False) + "\n"
+        self._pending_lines.setdefault(stream, []).append(line)
+
+    def _get_s3(self):
+        if self._s3 is not None:
+            return self._s3
+        if not _has_s3_credentials():
+            return None
+        try:
+            import boto3
+            self._s3 = boto3.client("s3", region_name=S3_REGION)
+            return self._s3
+        except Exception as e:
+            logger.debug("Spine S3 client unavailable: %s", e)
+            return None
+
+    def _read_local_stream_payload(
+        self,
+        stream: str,
+        exclude_lines: Optional[Iterable[str]] = None,
+    ) -> bytes:
+        path_attr = S3_PATH_ATTR_MAP.get(stream)
+        path = getattr(self, path_attr, None) if path_attr else None
+        if path is None:
+            return b""
+        try:
+            payload = Path(path).read_bytes()
+        except OSError:
+            return b""
+        return _drop_trailing_jsonl_lines(payload, exclude_lines or [])
+
+    def flush_pending_to_s3(self, max_lines_per_stream: int = 200) -> None:
+        """Flush queued records to S3 in bounded batches (heartbeat cadence)."""
+        if not self._pending_lines:
+            return
+        s3 = self._get_s3()
+        if not s3:
+            return
+
+        for stream, pending in list(self._pending_lines.items()):
+            if not pending:
+                continue
+            key = S3_KEY_MAP.get(stream)
+            if not key:
+                continue
+            batch_size = max(1, max_lines_per_stream)
+            batch = pending[:batch_size]
+            try:
+                try:
+                    resp = s3.get_object(Bucket=S3_BUCKET, Key=key)
+                    existing = resp["Body"].read()
+                except Exception as e:
+                    if _is_missing_s3_key_error(s3, e):
+                        existing = b""
+                    else:
+                        raise
+                pending_payload = "".join(batch).encode("utf-8")
+                local_payload = self._read_local_stream_payload(
+                    stream,
+                    exclude_lines=pending[len(batch):],
+                )
+                body = _merge_jsonl_payloads(
+                    existing,
+                    local_payload,
+                    pending_payload,
+                )
+                s3.put_object(
+                    Bucket=S3_BUCKET,
+                    Key=key,
+                    Body=body,
+                    ContentType="application/x-ndjson",
+                )
+                del pending[: len(batch)]
+            except Exception as e:
+                logger.debug("Spine S3 flush failed for %s: %s", stream, e)
 
     def append_dilemma(
         self,
@@ -122,7 +291,7 @@ class GovernanceEventSpine:
             "timestamp": timestamp,
             "payload": payload,
         }
-        _append_jsonl(self.internal_dilemmas_path, record)
+        self._append_stream(self.internal_dilemmas_path, record, "internal_dilemmas")
         self._append_stats("internal_dilemmas", dilemma_id, source=source)
         return DilemmaRef(
             dilemma_id=dilemma_id,
@@ -151,7 +320,7 @@ class GovernanceEventSpine:
             "timestamp": timestamp,
             "payload": advisory,
         }
-        _append_jsonl(self.oracle_advisories_path, record)
+        self._append_stream(self.oracle_advisories_path, record, "oracle_advisories")
         self._append_stats("oracle_advisories", advisory_id, source=source)
         return AdvisoryRef(
             advisory_id=advisory_id,
@@ -182,7 +351,9 @@ class GovernanceEventSpine:
             "timestamp": timestamp,
             "payload": resolution,
         }
-        _append_jsonl(self.synthesis_resolutions_path, record)
+        self._append_stream(
+            self.synthesis_resolutions_path, record, "synthesis_resolutions"
+        )
         self._append_stats("synthesis_resolutions", synthesis_id, source=source)
         return ResolutionRef(
             synthesis_id=synthesis_id,
@@ -213,7 +384,11 @@ class GovernanceEventSpine:
             "timestamp": timestamp,
             "payload": decision,
         }
-        _append_jsonl(self.synthesis_council_decisions_path, record)
+        self._append_stream(
+            self.synthesis_council_decisions_path,
+            record,
+            "synthesis_council_decisions",
+        )
         self._append_stats("synthesis_council_decisions", synthesis_id, source=source)
         return ResolutionRef(
             synthesis_id=synthesis_id,
@@ -242,7 +417,7 @@ class GovernanceEventSpine:
             "timestamp": timestamp,
             "payload": event,
         }
-        _append_jsonl(self.federation_events_path, record)
+        self._append_stream(self.federation_events_path, record, "federation_events")
         self._append_stats("federation_events", event_id, source=source)
         return FederationEventRef(
             federation_event_id=event_id,
@@ -275,7 +450,7 @@ class GovernanceEventSpine:
             "timestamp": timestamp,
             "payload": update,
         }
-        _append_jsonl(self.ark_updates_path, record)
+        self._append_stream(self.ark_updates_path, record, "ark_updates")
         self._append_stats("ark_updates", update_id, source=source)
         return ArkUpdateRef(
             ark_update_id=update_id,
@@ -285,7 +460,7 @@ class GovernanceEventSpine:
         )
 
     def append_living_axiom(self, entry: Dict[str, Any]) -> None:
-        _append_jsonl(self.living_axioms_path, entry)
+        self._append_stream(self.living_axioms_path, entry)
 
     def iter_oracle_advisories(self) -> Iterable[Dict[str, Any]]:
         return _read_jsonl(self.oracle_advisories_path)
@@ -303,5 +478,4 @@ class GovernanceEventSpine:
             "source": source,
             "timestamp": _utc_now(),
         }
-        _append_jsonl(self.federation_stats_path, record)
-
+        self._append_stream(self.federation_stats_path, record, "federation_stats")

elpidaapp/parliament_cycle_engine.py

@@ -2560,6 +2560,147 @@ class ParliamentCycleEngine:
         else:
             print("   📚 D14 restore: no prior constitutional axioms in S3 yet")
 
+    def _restore_step3_bottom_lane_memory(self) -> None:
+        """Restore Step-3 persistence files from federation/world S3 keys."""
+        s3_bridge = self._get_s3()
+        if s3_bridge is None:
+            logger.warning("Step3 restore skipped — S3 unavailable")
+            return
+
+        try:
+            from s3_bridge import BUCKET_BODY, REGION_BODY, BUCKET_WORLD, REGION_WORLD
+        except Exception as e:
+            logger.warning("Step3 restore skipped — cannot import S3 constants: %s", e)
+            return
+
+        body_s3 = s3_bridge._get_s3(REGION_BODY)
+        world_s3 = s3_bridge._get_s3(REGION_WORLD)
+        cache_dir = Path(__file__).resolve().parent.parent / "cache"
+        restore_targets = [
+            (
+                body_s3,
+                BUCKET_BODY,
+                "federation/contradiction_log.jsonl",
+                cache_dir / "contradiction_log.jsonl",
+            ),
+            (
+                body_s3,
+                BUCKET_BODY,
+                "federation/sacrifice_ledger.jsonl",
+                cache_dir / "governance_sacrifices.jsonl",
+            ),
+            (
+                world_s3,
+                BUCKET_WORLD,
+                "proposals/polis_civic_memory.json",
+                Path(__file__).resolve().parent.parent / "POLIS" / "polis_civic_memory.json",
+            ),
+        ]
+
+        spine = self._get_governance_event_spine()
+        if spine:
+            restore_targets.extend(
+                [
+                    (body_s3, BUCKET_BODY, "federation/internal_dilemmas.jsonl", spine.internal_dilemmas_path),
+                    (body_s3, BUCKET_BODY, "federation/oracle_advisories.jsonl", spine.oracle_advisories_path),
+                    (body_s3, BUCKET_BODY, "federation/synthesis_council_decisions.jsonl", spine.synthesis_council_decisions_path),
+                    (body_s3, BUCKET_BODY, "federation/synthesis_resolutions.jsonl", spine.synthesis_resolutions_path),
+                    (body_s3, BUCKET_BODY, "federation/federation_events.jsonl", spine.federation_events_path),
+                    (body_s3, BUCKET_BODY, "federation/ark_updates.jsonl", spine.ark_updates_path),
+                    (body_s3, BUCKET_BODY, "federation/federation_stats.jsonl", spine.federation_stats_path),
+                ]
+            )
+
+        restored = 0
+        for client, bucket, key, local_path in restore_targets:
+            if client is None:
+                continue
+            try:
+                resp = client.get_object(Bucket=bucket, Key=key)
+                payload = resp["Body"].read()
+                repair_s3 = False
+
+                if local_path.suffix == ".jsonl" and local_path.exists():
+                    local_payload = local_path.read_bytes()
+                    if local_payload != payload:
+                        payload, repair_s3 = self._merge_step3_jsonl_restore(
+                            payload, local_payload
+                        )
+                elif local_path.exists():
+                    local_payload = local_path.read_bytes()
+                    if (
+                        local_payload != payload
+                        and self._step3_local_is_newer(local_path, resp.get("LastModified"))
+                    ):
+                        payload = local_payload
+                        repair_s3 = True
+
+                local_path.parent.mkdir(parents=True, exist_ok=True)
+                local_path.write_bytes(payload)
+                if repair_s3:
+                    self._repair_step3_s3_object(client, bucket, key, payload, local_path)
+                restored += 1
+            except Exception:
+                continue
+
+        if restored:
+            print(f"   🧷 Step3 bottom-lane restore: {restored} file(s) synced from S3")
+
+    @staticmethod
+    def _merge_step3_jsonl_restore(
+        s3_payload: bytes,
+        local_payload: bytes,
+    ) -> Tuple[bytes, bool]:
+        """Merge S3 and local JSONL bodies without dropping local-only rows."""
+        merged_lines: List[str] = []
+        seen = set()
+        for payload in (s3_payload, local_payload):
+            text = payload.decode("utf-8", errors="replace")
+            for line in text.splitlines():
+                if not line.strip() or line in seen:
+                    continue
+                merged_lines.append(line)
+                seen.add(line)
+        merged = (
+            "\n".join(merged_lines) + ("\n" if merged_lines else "")
+        ).encode("utf-8")
+        return merged, merged != s3_payload
+
+    @staticmethod
+    def _step3_local_is_newer(local_path: Path, s3_last_modified: Any) -> bool:
+        if s3_last_modified is None:
+            return False
+        try:
+            local_mtime = datetime.fromtimestamp(local_path.stat().st_mtime, timezone.utc)
+            if s3_last_modified.tzinfo is None:
+                s3_last_modified = s3_last_modified.replace(tzinfo=timezone.utc)
+            return local_mtime > s3_last_modified
+        except Exception:
+            return False
+
+    @staticmethod
+    def _repair_step3_s3_object(
+        client: Any,
+        bucket: str,
+        key: str,
+        payload: bytes,
+        local_path: Path,
+    ) -> None:
+        content_type = (
+            "application/x-ndjson"
+            if local_path.suffix == ".jsonl"
+            else "application/json; charset=utf-8"
+        )
+        try:
+            client.put_object(
+                Bucket=bucket,
+                Key=key,
+                Body=payload,
+                ContentType=content_type,
+            )
+        except Exception as e:
+            logger.debug("Step3 restore S3 repair failed for %s: %s", key, e)
+
     def _emit_d16_execution(self, exec_type: str, proposal: str,
                              meta: Optional[Dict] = None) -> None:
         """
@@ -2736,6 +2877,26 @@ class ParliamentCycleEngine:
         - approval_rate, veto_exercised, axiom_frequency
         - d15_broadcast_count
         """
+        spine = self._get_governance_event_spine()
+        if spine:
+            try:
+                spine.flush_pending_to_s3()
+            except Exception as e:
+                logger.debug("GovernanceEventSpine flush skipped: %s", e)
+        if self._contradiction_log:
+            try:
+                self._contradiction_log.flush_pending_to_s3()
+            except Exception as e:
+                logger.debug("ContradictionLog flush skipped: %s", e)
+        if self._gov_sacrifice_tracker:
+            try:
+                self._gov_sacrifice_tracker.flush_pending_to_s3()
+            except Exception as e:
+                logger.debug(
+                    "GovernanceSacrificeTracker flush skipped: %s",
+                    e,
+                )
+
         watch = self._watch.current()
         heartbeat = {
             "source": "BODY",
@@ -3953,6 +4114,9 @@ class ParliamentCycleEngine:
         # across all previous spirals (cross-spiral awareness, mirrors MIND Ark fix).
         self._restore_d15_broadcast_state()
 
+        # Step 3 persistence — restore bottom-lane runtime files from S3.
+        self._restore_step3_bottom_lane_memory()
+
         # D0↔D11 Body Bridge — restore arc coherence state from cache before first cycle.
         try:
             from elpidaapp.domain_0_11_connector_body import get_body_connector

elpidaapp/polis_bridge.py

@@ -35,6 +35,7 @@ Usage in parliament_cycle_engine.py::
 
 import json
 import logging
+import os
 from pathlib import Path
 from datetime import datetime, timezone
 from typing import Dict, List, Optional, Any, Tuple
@@ -65,11 +66,26 @@ for p_ax, a_list in P_TO_A_MAP.items():
 
 # POLIS civic memory default path (relative to repo root)
 DEFAULT_CIVIC_MEMORY = "POLIS/polis_civic_memory.json"
+POLIS_S3_BUCKET = os.environ.get("POLIS_S3_BUCKET", "elpida-external-interfaces")
+POLIS_S3_REGION = os.environ.get("POLIS_S3_REGION", "eu-north-1")
+POLIS_S3_KEY = os.environ.get("POLIS_S3_KEY", "proposals/polis_civic_memory.json")
+_MISSING_S3_KEY_CODES = {"NoSuchKey", "404", "NotFound"}
 
 # How many cycles between POLIS bridge consultations (Fibonacci)
 POLIS_CYCLE_INTERVAL = 34
 
 
+def _is_missing_s3_key_error(s3, exc: Exception) -> bool:
+    no_such_key = getattr(getattr(s3, "exceptions", None), "NoSuchKey", None)
+    if isinstance(no_such_key, type) and isinstance(exc, no_such_key):
+        return True
+    response = getattr(exc, "response", None)
+    if isinstance(response, dict):
+        code = response.get("Error", {}).get("Code")
+        return str(code) in _MISSING_S3_KEY_CODES
+    return False
+
+
 class PolisBridge:
     """
     Bridge between POLIS civic contradictions and Elpida parliament.
@@ -100,6 +116,7 @@ class PolisBridge:
 
         self._last_processed_id: Optional[str] = None
         self._processed_ids: set = set()
+        self._s3 = None
 
         if self._path.exists():
             logger.info(
@@ -116,6 +133,14 @@ class PolisBridge:
 
     def _load_memory(self) -> Dict[str, Any]:
         """Load the full POLIS civic memory JSON."""
+        s3_memory = self._load_memory_from_s3()
+        if s3_memory is not None:
+            return s3_memory
+
+        return self._load_memory_from_local()
+
+    def _load_memory_from_local(self) -> Dict[str, Any]:
+        """Load POLIS civic memory from the local filesystem."""
         if not self._path.exists():
             return {}
         try:
@@ -125,6 +150,101 @@ class PolisBridge:
             logger.warning("Cannot read POLIS civic memory: %s", e)
             return {}
 
+    def _get_s3(self):
+        if self._s3 is not None:
+            return self._s3
+        sink_enabled = os.environ.get("POLIS_S3_SYNC", "1").strip().lower()
+        if sink_enabled in {"0", "false", "no", "off", ""}:
+            return None
+        if not (
+            os.environ.get("AWS_ACCESS_KEY_ID")
+            and os.environ.get("AWS_SECRET_ACCESS_KEY")
+        ):
+            return None
+        try:
+            import boto3
+            self._s3 = boto3.client("s3", region_name=POLIS_S3_REGION)
+            return self._s3
+        except Exception as e:
+            logger.debug("POLIS S3 client unavailable: %s", e)
+            return None
+
+    def _load_memory_from_s3(self) -> Optional[Dict[str, Any]]:
+        s3 = self._get_s3()
+        if not s3:
+            return None
+        try:
+            resp = s3.get_object(Bucket=POLIS_S3_BUCKET, Key=POLIS_S3_KEY)
+            payload_bytes = resp["Body"].read()
+            local_payload = self._read_local_memory_bytes()
+            if (
+                local_payload is not None
+                and local_payload != payload_bytes
+                and self._local_memory_newer_than(resp.get("LastModified"))
+            ):
+                try:
+                    local_memory = json.loads(local_payload.decode("utf-8"))
+                    self._save_memory_to_s3(local_memory)
+                    return local_memory
+                except (json.JSONDecodeError, UnicodeDecodeError):
+                    pass
+
+            payload = json.loads(payload_bytes.decode("utf-8"))
+            # Keep local seed in sync for fallback/startup continuity.
+            try:
+                self._path.parent.mkdir(parents=True, exist_ok=True)
+                with open(self._path, "w", encoding="utf-8") as f:
+                    json.dump(payload, f, indent=2, ensure_ascii=False)
+            except Exception:
+                pass
+            return payload
+        except (json.JSONDecodeError, UnicodeDecodeError) as e:
+            logger.warning("POLIS S3 memory decode failed; falling back to local: %s", e)
+            return None
+        except Exception as e:
+            if _is_missing_s3_key_error(s3, e):
+                return None
+            logger.warning("POLIS S3 memory read failed; refusing stale local fallback: %s", e)
+            return {}
+
+    def _read_local_memory_bytes(self) -> Optional[bytes]:
+        if not self._path.exists():
+            return None
+        try:
+            return self._path.read_bytes()
+        except OSError:
+            return None
+
+    def _local_memory_newer_than(self, s3_last_modified: Any) -> bool:
+        if s3_last_modified is None:
+            return False
+        try:
+            local_mtime = datetime.fromtimestamp(
+                self._path.stat().st_mtime,
+                timezone.utc,
+            )
+            if s3_last_modified.tzinfo is None:
+                s3_last_modified = s3_last_modified.replace(tzinfo=timezone.utc)
+            return local_mtime > s3_last_modified
+        except Exception:
+            return False
+
+    def _save_memory_to_s3(self, memory: Dict[str, Any]) -> bool:
+        s3 = self._get_s3()
+        if not s3:
+            return False
+        try:
+            s3.put_object(
+                Bucket=POLIS_S3_BUCKET,
+                Key=POLIS_S3_KEY,
+                Body=json.dumps(memory, ensure_ascii=False, indent=2).encode("utf-8"),
+                ContentType="application/json; charset=utf-8",
+            )
+            return True
+        except Exception as e:
+            logger.debug("POLIS S3 write failed: %s", e)
+            return False
+
     def get_held_contradictions(self) -> List[Dict[str, Any]]:
         """Return all HELD (unresolved) contradictions from POLIS memory."""
         memory = self._load_memory()
@@ -340,6 +460,7 @@ class PolisBridge:
             # Write back to file
             with open(self._path, "w", encoding="utf-8") as f:
                 json.dump(memory, f, indent=2, ensure_ascii=False)
+            self._save_memory_to_s3(memory)
 
             logger.info(
                 "POLIS branch written: %s → %s (%s)",

elpidaapp/sacrifice_tracker.py

@@ -30,9 +30,10 @@ from __future__ import annotations
 
 import json
 import logging
+import os
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Iterable, List, Optional
 
 logger = logging.getLogger("elpidaapp.sacrifice_tracker")
 
@@ -213,6 +214,63 @@ def create_sacrifice_tracker(
 # ---------------------------------------------------------------------------
 
 _GOV_SACRIFICE_LOG = Path(__file__).resolve().parent.parent / "cache" / "governance_sacrifices.jsonl"
+_GOV_SACRIFICE_S3_BUCKET = os.environ.get("SACRIFICE_S3_BUCKET", "elpida-body-evolution")
+_GOV_SACRIFICE_S3_REGION = os.environ.get("SACRIFICE_S3_REGION", "eu-north-1")
+_GOV_SACRIFICE_S3_KEY = os.environ.get("SACRIFICE_S3_KEY", "federation/sacrifice_ledger.jsonl")
+_MISSING_S3_KEY_CODES = {"NoSuchKey", "404", "NotFound"}
+
+
+def _is_missing_s3_key_error(s3, exc: Exception) -> bool:
+    no_such_key = getattr(getattr(s3, "exceptions", None), "NoSuchKey", None)
+    if isinstance(no_such_key, type) and isinstance(exc, no_such_key):
+        return True
+    response = getattr(exc, "response", None)
+    if isinstance(response, dict):
+        code = response.get("Error", {}).get("Code")
+        return str(code) in _MISSING_S3_KEY_CODES
+    return False
+
+
+def _has_s3_credentials() -> bool:
+    return bool(
+        os.environ.get("AWS_ACCESS_KEY_ID")
+        and os.environ.get("AWS_SECRET_ACCESS_KEY")
+    )
+
+
+def _merge_jsonl_payloads(*payloads: bytes) -> bytes:
+    """Merge JSONL bodies without dropping rows present only in local cache."""
+    merged_lines: List[str] = []
+    seen = set()
+    for payload in payloads:
+        text = payload.decode("utf-8", errors="replace")
+        for line in text.splitlines():
+            if not line.strip() or line in seen:
+                continue
+            merged_lines.append(line)
+            seen.add(line)
+    return ("\n".join(merged_lines) + ("\n" if merged_lines else "")).encode("utf-8")
+
+
+def _drop_trailing_jsonl_lines(payload: bytes, lines_to_drop: Iterable[str]) -> bytes:
+    drop_counts: Dict[str, int] = {}
+    for line in lines_to_drop:
+        key = line.rstrip("\n")
+        drop_counts[key] = drop_counts.get(key, 0) + 1
+    if not drop_counts:
+        return payload
+
+    text = payload.decode("utf-8", errors="replace")
+    kept_reversed: List[str] = []
+    for line in reversed(text.splitlines()):
+        count = drop_counts.get(line, 0)
+        if count:
+            drop_counts[line] = count - 1
+            continue
+        kept_reversed.append(line)
+    kept = list(reversed(kept_reversed))
+    return ("\n".join(kept) + ("\n" if kept else "")).encode("utf-8")
+
 
 # Mapping: sacrifice type → (axiom_cost, axiom_served)
 # What value was traded away, and what value was protected.
@@ -237,6 +295,48 @@ class GovernanceSacrificeTracker:
     def __init__(self):
         self._count: int = 0
         self._type_counts: Dict[str, int] = {}
+        self._s3 = None
+        self._pending_lines: List[str] = []
+        self._hydrate_from_disk()
+
+    def _hydrate_from_disk(self) -> None:
+        """Initialize counters from the restored governance sacrifice ledger."""
+        if not _GOV_SACRIFICE_LOG.exists():
+            return
+        records = 0
+        max_number = 0
+        type_counts: Dict[str, int] = {}
+        try:
+            with open(_GOV_SACRIFICE_LOG, "r", encoding="utf-8") as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    try:
+                        entry = json.loads(line)
+                    except json.JSONDecodeError:
+                        continue
+                    records += 1
+                    try:
+                        max_number = max(
+                            max_number,
+                            int(entry.get("sacrifice_number", 0)),
+                        )
+                    except (TypeError, ValueError):
+                        pass
+                    sacrifice_type = entry.get("type")
+                    if sacrifice_type:
+                        type_counts[sacrifice_type] = (
+                            type_counts.get(sacrifice_type, 0) + 1
+                        )
+            self._count = max(records, max_number)
+            self._type_counts = type_counts
+            logger.info(
+                "GovernanceSacrificeTracker: loaded %d existing records from %s",
+                records, _GOV_SACRIFICE_LOG,
+            )
+        except Exception as e:
+            logger.warning("GovernanceSacrificeTracker: could not read ledger: %s", e)
 
     def record(
         self,
@@ -285,6 +385,7 @@ class GovernanceSacrificeTracker:
                 f.write(json.dumps(entry, ensure_ascii=False) + "\n")
         except Exception as e:
             logger.debug("Gov sacrifice persist failed: %s", e)
+        self._queue_for_s3(entry)
 
         return entry
 
@@ -294,3 +395,80 @@ class GovernanceSacrificeTracker:
             "total": self._count,
             "type_counts": dict(self._type_counts),
         }
+
+    def _get_s3(self):
+        if self._s3 is not None:
+            return self._s3
+        sink_enabled = os.environ.get("SACRIFICE_S3_SINK", "1").strip().lower()
+        if sink_enabled in {"0", "false", "no", "off", ""}:
+            return None
+        if not _has_s3_credentials():
+            return None
+        try:
+            import boto3
+            self._s3 = boto3.client("s3", region_name=_GOV_SACRIFICE_S3_REGION)
+            return self._s3
+        except Exception as e:
+            logger.debug("Sacrifice S3 client unavailable: %s", e)
+            return None
+
+    def _queue_for_s3(self, entry: Dict[str, Any]) -> None:
+        sink_enabled = os.environ.get("SACRIFICE_S3_SINK", "1").strip().lower()
+        if sink_enabled in {"0", "false", "no", "off", ""}:
+            return
+        if getattr(self, "_s3", None) is None and not _has_s3_credentials():
+            return
+        if not hasattr(self, "_pending_lines"):
+            self._pending_lines = []
+        line = json.dumps(entry, ensure_ascii=False) + "\n"
+        self._pending_lines.append(line)
+
+    def flush_pending_to_s3(self, max_lines: int = 200) -> None:
+        """Flush queued sacrifice records to S3 on heartbeat cadence."""
+        pending = getattr(self, "_pending_lines", [])
+        if not pending:
+            return
+        s3 = self._get_s3()
+        if not s3:
+            return
+        batch_size = max(1, max_lines)
+        batch = pending[:batch_size]
+        try:
+            try:
+                resp = s3.get_object(
+                    Bucket=_GOV_SACRIFICE_S3_BUCKET,
+                    Key=_GOV_SACRIFICE_S3_KEY,
+                )
+                existing = resp["Body"].read()
+            except Exception as e:
+                if _is_missing_s3_key_error(s3, e):
+                    existing = b""
+                else:
+                    raise
+            try:
+                local_payload = _GOV_SACRIFICE_LOG.read_bytes()
+            except OSError:
+                local_payload = b""
+            local_payload = _drop_trailing_jsonl_lines(
+                local_payload,
+                pending[len(batch):],
+            )
+            pending_payload = "".join(batch).encode("utf-8")
+            s3.put_object(
+                Bucket=_GOV_SACRIFICE_S3_BUCKET,
+                Key=_GOV_SACRIFICE_S3_KEY,
+                Body=_merge_jsonl_payloads(
+                    existing,
+                    local_payload,
+                    pending_payload,
+                ),
+                ContentType="application/x-ndjson",
+            )
+            del pending[: len(batch)]
+        except Exception as e:
+            logger.debug("Sacrifice S3 flush failed: %s", e)
+
+    def _append_to_s3(self, entry: Dict[str, Any]) -> None:
+        """Compatibility path; runtime uses heartbeat flush."""
+        self._queue_for_s3(entry)
+        self.flush_pending_to_s3(max_lines=1)

elpidaapp/scanner.py

@@ -2,7 +2,7 @@
 """
 Problem Scanner — Autonomous dilemma finder for ElpidaApp.
 
-Uses Perplexity (D13 — Archive/External Interface) to find
+Uses live web grounding (Brave/DDG/Wikipedia) to find
 real-world dilemmas worth analyzing, then structures them into
 properly-formatted problem statements for the Divergence Engine.
 
@@ -56,9 +56,9 @@ SCAN_TOPICS = [
 
 class ProblemScanner:
     """
-    Finds real-world policy dilemmas by searching the web via
-    DuckDuckGo (free, no API key) and structuring results with
-    a free LLM (Groq).
+    Finds real-world policy dilemmas by searching live web sources
+    (Brave API when available, then DuckDuckGo, then Wikipedia)
+    and structuring results with an LLM.
     """
 
     def __init__(
@@ -88,7 +88,7 @@ class ProblemScanner:
             scan_topic = topic or self._pick_topic(i)
             print(f"\n🔍 Scanning: {scan_topic}...")
 
-            # Step 1: Ask Perplexity for current real-world dilemmas (with citations)
+            # Step 1: collect current real-world dilemmas from live web context
             research = self._research_dilemmas(scan_topic)
             raw_dilemmas = research.get("text") if isinstance(research, dict) else research
             citations = research.get("citations", []) if isinstance(research, dict) else []
@@ -99,7 +99,7 @@ class ProblemScanner:
             # Step 2: Structure into a proper problem statement
             structured = self._structure_problem(raw_dilemmas, scan_topic)
             if structured:
-                # Build source list from Perplexity citations
+                # Build source list from collected citations
                 sources = []
                 for url in citations:
                     if isinstance(url, str) and url.startswith("http"):
@@ -157,40 +157,47 @@ class ProblemScanner:
 
     def _research_dilemmas(self, topic: str) -> Dict[str, Any]:
         """
-        Use DuckDuckGo web search + Groq to find real-world active
-        dilemmas with genuine competing interests. Falls back to Wikipedia
-        when DDG returns empty (HF Space frequently throttles DDG).
+        Use live web search + Groq to find real-world active dilemmas
+        with genuine competing interests. Search order:
+        Brave API (if BRAVE_API_KEY) -> DuckDuckGo -> Wikipedia.
 
         Returns {"text": str, "citations": list[str]}.
         """
-        from elpidaapp.domain_grounding import _rate_limit, _search_wikipedia
+        from elpidaapp.domain_grounding import (
+            _rate_limit,
+            _search_brave,
+            _search_wikipedia,
+        )
 
-        # ── Step 1: free web search via DDG ──
+        # ── Step 1: live web search (Brave -> DDG -> Wikipedia fallback) ──
         _rate_limit()
         query = f"{topic} policy dilemma controversy 2026"
-        raw_results = []
-        try:
-            from ddgs import DDGS
-            ddgs = DDGS()
-            raw_results = list(ddgs.text(query, max_results=6))
-        except Exception:
-            pass
-
-        # ── Step 1b: Wikipedia fallback when DDG returns empty ──
-        # Mirrors domain_grounding.ground_query's two-tier strategy.
+        raw_results = [
+            {
+                "title": r.get("title", ""),
+                "body": r.get("body", ""),
+                "href": r.get("href", ""),
+            }
+            for r in _search_brave(query, max_results=6)
+        ]
+
+        # DDG fallback if Brave unavailable/empty
+        if not raw_results:
+            try:
+                from ddgs import DDGS
+                ddgs = DDGS()
+                raw_results = list(ddgs.text(query, max_results=6))
+            except Exception:
+                pass
+
+        # Wikipedia fallback when web search returns empty.
         if not raw_results:
             wiki_results = _search_wikipedia(topic, max_results=5)
-            # Wikipedia API returns {title, body}; normalize to ddgs shape so
-            # the citation/snippet code below works unchanged.
             for r in wiki_results:
-                title = r.get("title", "")
                 raw_results.append({
-                    "title": title,
+                    "title": r.get("title", ""),
                     "body": r.get("body", ""),
-                    "href": (
-                        f"https://en.wikipedia.org/wiki/{title.replace(' ', '_')}"
-                        if title else ""
-                    ),
+                    "href": r.get("href", ""),
                 })
 
         # Extract URLs as citations and build context

elpidaapp/ui.py

@@ -988,7 +988,13 @@ with tab_audit:
             baseline_provider=_baseline,
         )
         with st.spinner(f"Running {preset_choice} analysis across {len(_domain_ids)} domains..."):
-            result = engine.analyze(problem)
+            _audit_out = (
+                Path(__file__).resolve().parent.parent
+                / "cache"
+                / "audits"
+                / f"audit_{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}.json"
+            )
+            result = engine.analyze(problem, save_to=str(_audit_out))
 
         # Resume BODY loop
         try:
@@ -1116,7 +1122,7 @@ with tab_scanner:
     st.markdown("### Scanner — Problem Discovery")
     st.markdown("""
     <div class="mode-intro">
-    Enter a topic or let Elpida choose. D13 (Archive / Perplexity) finds current
+    Enter a topic or let Elpida choose. D13 live grounding (Brave/DDG/Wikipedia) finds current
     real-world problems, then the Divergence Engine runs multi-domain analysis.
     </div>
     """, unsafe_allow_html=True)