feat: daily CNPG Option A backup shell activity (RAILIANCE-WP-0016)

Add cnpg_option_a_backup resolver, disabled ActivityDefinition, ESO
manifest, worker kubeconfig hostPath, databases RBAC, and unit tests.
Enable after ESO token re-mint and host kubeconfig wiring.

src/activity_core/context_resolvers/cnpg_backup.py

@@ -0,0 +1,96 @@
+"""CNPG Option A backup shell context query (RAILIANCE-WP-0016)."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import subprocess
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_SCRIPT = Path("/opt/railiance-platform/tools/cmd/cnpg-option-a-backup")
+_DEFAULT_TIMEOUT_SECONDS = 7200
+
+
+def cnpg_option_a_backup(params: dict[str, Any]) -> dict[str, Any]:
+    """Run multi-host Option A backup CLI and return its JSON summary."""
+    script = Path(str(params.get("backup_script", _DEFAULT_SCRIPT))).expanduser()
+    if not script.is_file():
+        raise FileNotFoundError(f"cnpg_option_a_backup script not found: {script}")
+
+    dry_run = bool(params.get("dry_run", False))
+    timeout = float(params.get("timeout_seconds", _DEFAULT_TIMEOUT_SECONDS))
+    targets = params.get("targets")  # optional CSV
+
+    env = os.environ.copy()
+    # Ensure vendor bin (age/kubectl) is visible when hostPath is mounted.
+    vendor = Path("/opt/railiance-platform/tools/vendor/bin")
+    if vendor.is_dir():
+        env["PATH"] = f"{vendor}:{env.get('PATH', '')}"
+
+    if dry_run:
+        env["RAILIANCE_BACKUP_DRY_RUN"] = "1"
+    else:
+        env["RAILIANCE_BACKUP_DRY_RUN"] = "0"
+
+    # Prefer env already injected via actcore-runtime-secret (ESO).
+    for src, dst in (
+        ("NC_WEBDAV_TOKEN", "RAILIANCE_BACKUP_NC_TOKEN"),
+        ("NC_WEBDAV_URL", "RAILIANCE_BACKUP_NC_WEBDAV_URL"),
+        ("AGE_PUBLIC_KEY", "RAILIANCE_BACKUP_AGE_PUBLIC_KEY"),
+    ):
+        if env.get(src) and not env.get(dst):
+            env[dst] = env[src]
+
+    if params.get("kubeconfig_core"):
+        env["KUBECONFIG_CORE"] = str(params["kubeconfig_core"])
+    if params.get("kubeconfig_r01"):
+        env["KUBECONFIG_R01"] = str(params["kubeconfig_r01"])
+
+    # Optional hostPath kubeconfigs mounted into the worker.
+    if not env.get("KUBECONFIG_CORE") and Path("/kube/config").is_file():
+        env["KUBECONFIG_CORE"] = "/kube/config"
+    if not env.get("KUBECONFIG_R01") and Path("/kube/config-hosteurope").is_file():
+        env["KUBECONFIG_R01"] = "/kube/config-hosteurope"
+    if not env.get("KUBECONFIG_R01") and Path("/kube/config").is_file():
+        # Single-cluster worker host: treat as railiance01.
+        env.setdefault("KUBECONFIG_R01", "/kube/config")
+
+    if targets:
+        env["CNPG_BACKUP_TARGETS"] = str(targets)
+
+    cmd = [str(script)]
+    completed = subprocess.run(
+        cmd,
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+        check=False,
+        env=env,
+    )
+    stdout = (completed.stdout or "").strip()
+    stderr = (completed.stderr or "").strip()
+    if completed.returncode not in (0, 2) and not stdout:
+        raise RuntimeError(
+            f"cnpg_option_a_backup failed (exit {completed.returncode}): {stderr[:500]}"
+        )
+
+    try:
+        summary = json.loads(stdout)
+    except json.JSONDecodeError as exc:
+        raise RuntimeError(
+            f"cnpg_option_a_backup returned invalid JSON: {exc}; stderr={stderr[:300]}"
+        ) from exc
+
+    if completed.returncode not in (0, 2):
+        summary.setdefault("errors", []).append(
+            f"script_exit_code:{completed.returncode}"
+        )
+    summary["kind"] = "cnpg_option_a_backup"
+    summary["script_exit_code"] = completed.returncode
+    if stderr:
+        summary["log_tail"] = stderr[-1500:]
+    return summary

src/activity_core/context_resolvers/reuse_surface.py

@@ -22,6 +22,7 @@ import yaml
 
 from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY, ContextResolver
 from activity_core.context_resolvers.forgejo_prune import forgejo_package_prune
+from activity_core.context_resolvers.cnpg_backup import cnpg_option_a_backup
 from activity_core.context_resolvers.kaizen import KaizenContextResolver
 from activity_core.context_resolvers.state_hub import StateHubContextResolver
 
@@ -512,6 +513,8 @@ class ShellContextResolver(ContextResolver):
             return reuse_surface_report_gaps(params)
         if query == "forgejo_package_prune":
             return forgejo_package_prune(params)
+        if query == "cnpg_option_a_backup":
+            return cnpg_option_a_backup(params)
         return KaizenContextResolver().resolve(query, event, params)
 
 

src/activity_core/ops_console.py

@@ -39,6 +39,7 @@ _AUDIT_BUFFER: deque[dict[str, Any]] = deque(maxlen=100)
 
 SIDE_EFFECT_MARKERS = (
     "forgejo_package_prune",
+    "cnpg_option_a_backup",
     "apply: true",
     '"apply": true',
     "'apply': true",

src/activity_core/ops_evidence_sinks.py

@@ -139,6 +139,10 @@ def _post_state_hub_progress(
         compact = probe_result
         summary = _forgejo_package_prune_summary_text(probe_result)
         source_type = "shell"
+    elif probe_result.get("kind") == "cnpg_option_a_backup":
+        compact = probe_result
+        summary = _cnpg_option_a_backup_summary_text(probe_result)
+        source_type = "shell"
     elif probe_result.get("checks") is not None:
         compact = probe_result
         summary = _phase5_summary_text(probe_result)
@@ -552,6 +556,18 @@ def _forgejo_package_prune_summary_text(result: dict[str, Any]) -> str:
     )
 
 
+def _cnpg_option_a_backup_summary_text(result: dict[str, Any]) -> str:
+    overall = result.get("overall", "?")
+    dumped = result.get("dumped", 0)
+    uploaded = result.get("uploaded", 0)
+    failed = result.get("failed", 0)
+    dry = "dry-run" if result.get("dry_run") else "live"
+    return (
+        f"CNPG Option A backup ({dry}): overall={overall}; "
+        f"dumped={dumped} uploaded={uploaded} failed={failed}"
+    )
+
+
 def _legacy_meter_summary_text(result: dict[str, Any]) -> str:
     candidates = result.get("retirement_candidate_count", 0)
     calls = result.get("window_legacy_calls", 0)