feat(automation): weekly Forgejo package prune shell resolver (WP-0020)

Add forgejo_package_prune query, weekly activity-definition (disabled),
State Hub evidence sink support for shell sources, and runbook notes.

src/activity_core/context_resolvers/forgejo_prune.py

@@ -0,0 +1,62 @@
+"""Forgejo package retention prune shell context query."""
+
+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.home() / "railiance-platform" / "tools/cmd/forgejo-package-prune"
+_DEFAULT_TIMEOUT_SECONDS = 900
+
+
+def forgejo_package_prune(params: dict[str, Any]) -> dict[str, Any]:
+    """Run the platform prune script and return its JSON summary."""
+    script = Path(str(params.get("prune_script", _DEFAULT_SCRIPT))).expanduser()
+    if not script.is_file():
+        raise FileNotFoundError(f"forgejo_package_prune script not found: {script}")
+
+    max_versions = int(params.get("max_versions", 3))
+    apply = bool(params.get("apply", False))
+    timeout = float(params.get("timeout_seconds", _DEFAULT_TIMEOUT_SECONDS))
+
+    cmd = [str(script), f"--max-versions={max_versions}"]
+    if apply:
+        cmd.append("--apply")
+
+    env = os.environ.copy()
+    completed = subprocess.run(
+        cmd,
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+        check=False,
+        env=env,
+    )
+    stdout = (completed.stdout or "").strip()
+    if completed.returncode != 0 and not stdout:
+        stderr = (completed.stderr or "").strip()
+        snippet = stderr[:500]
+        raise RuntimeError(
+            f"forgejo_package_prune failed (exit {completed.returncode}): {snippet}"
+        )
+
+    try:
+        summary = json.loads(stdout)
+    except json.JSONDecodeError as exc:
+        stderr = (completed.stderr or "").strip()
+        raise RuntimeError(
+            f"forgejo_package_prune returned invalid JSON: {exc}; stderr={stderr[:300]}"
+        ) from exc
+
+    if completed.returncode != 0:
+        summary.setdefault("errors", []).append(
+            f"script_exit_code:{completed.returncode}"
+        )
+    summary["kind"] = "forgejo_package_prune"
+    return summary

src/activity_core/context_resolvers/reuse_surface.py

@@ -21,6 +21,7 @@ import httpx
 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.kaizen import KaizenContextResolver
 from activity_core.context_resolvers.state_hub import StateHubContextResolver
 
@@ -509,6 +510,8 @@ class ShellContextResolver(ContextResolver):
     def resolve(self, query: str, event: Any, params: dict[str, Any]) -> dict[str, Any]:
         if query == "reuse_surface_report_gaps":
             return reuse_surface_report_gaps(params)
+        if query == "forgejo_package_prune":
+            return forgejo_package_prune(params)
         return KaizenContextResolver().resolve(query, event, params)
 
 

src/activity_core/ops_evidence_sinks.py

@@ -40,7 +40,7 @@ def persist_ops_inventory_evidence(payload: dict[str, Any]) -> list[dict[str, An
         if not isinstance(source, dict):
             continue
         source_type = source.get("type")
-        if source_type not in {"ops-inventory", "state-hub", "core-hub"}:
+        if source_type not in {"ops-inventory", "state-hub", "core-hub", "shell"}:
             continue
 
         params = source.get("params") or {}
@@ -135,6 +135,10 @@ def _post_state_hub_progress(
         compact = probe_result
         summary = _legacy_meter_summary_text(probe_result)
         source_type = "state-hub"
+    elif probe_result.get("kind") == "forgejo_package_prune":
+        compact = probe_result
+        summary = _forgejo_package_prune_summary_text(probe_result)
+        source_type = "shell"
     elif probe_result.get("checks") is not None:
         compact = probe_result
         summary = _phase5_summary_text(probe_result)
@@ -534,6 +538,20 @@ def _core_hub_stabilization_summary_text(result: dict[str, Any]) -> str:
     )
 
 
+def _forgejo_package_prune_summary_text(result: dict[str, Any]) -> str:
+    deleted = result.get("deleted_count", 0)
+    candidates = result.get("candidate_count", 0)
+    protected = result.get("skipped_protected_count", 0)
+    apply = result.get("apply", False)
+    mode = "apply" if apply else "dry-run"
+    errors = result.get("errors") or []
+    error_note = f"; {len(errors)} error(s)" if errors else ""
+    return (
+        f"Forgejo package prune ({mode}): {deleted} deleted, "
+        f"{candidates} candidate(s), {protected} protected skip(s){error_note}"
+    )
+
+
 def _legacy_meter_summary_text(result: dict[str, Any]) -> str:
     candidates = result.get("retirement_candidate_count", 0)
     calls = result.get("window_legacy_calls", 0)