feat: Phase 5 stabilization schedules and custodian path URIs

Add phase5_stabilization_check State Hub resolver with progress evidence
sinks, schedule projections for daily and closeout checks, custodian:// and
activity-core:// runtime path resolution, and Railiance mounts under /var/custodian.

src/activity_core/activities.py

@@ -156,7 +156,14 @@ async def resolve_context(
         bind_key = raw_bind.removeprefix("context.") if raw_bind.startswith("context.") else raw_bind
 
         if source_type == "static":
-            snapshot[bind_key] = source.get("config", {}).get("value")
+            value = source.get("config", {}).get("value")
+            if isinstance(value, str) and (
+                value.startswith("custodian://") or value.startswith("activity-core://")
+            ):
+                from activity_core.runtime_paths import resolve_runtime_path
+
+                value = str(resolve_runtime_path(value))
+            snapshot[bind_key] = value
             continue
 
         resolver_cls = CONTEXT_RESOLVER_REGISTRY.get(source_type)

src/activity_core/context_resolvers/state_hub.py

@@ -13,6 +13,7 @@ Supported queries:
   - daily_triage_digest: curated scalar JSON digest for daily WSJF triage
   - recently_on_scope_hourly: POST {STATE_HUB_URL}/recently-on-scope/hourly
   - consistency_sweep_remote_all: POST {STATE_HUB_URL}/consistency/sweep/remote-all
+  - phase5_stabilization_check: hub-visible Phase 5 stabilization gates
 
 No caching — state hub data is live operational state and must not be stale
 within a single workflow run.
@@ -134,9 +135,159 @@ class StateHubContextResolver(ContextResolver):
                 timeout=_SWEEP_TIMEOUT_SECONDS,
             )
             return _validate_consistency_sweep_remote_all(result)
+        if query == "phase5_stabilization_check":
+            return _phase5_stabilization_check(params)
         return {}
 
 
+def _parse_iso_datetime(raw: Any) -> datetime | None:
+    if not raw:
+        return None
+    text = str(raw).replace("Z", "+00:00")
+    try:
+        dt = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt
+
+
+def _utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _phase5_stabilization_check(params: dict[str, Any]) -> dict[str, Any]:
+    closeout = bool(params.get("closeout"))
+    now = _utc_now()
+    window_start = _parse_iso_datetime(
+        params.get("window_start", "2026-07-06T17:35:00+00:00")
+    )
+    window_end = _parse_iso_datetime(
+        params.get("window_end", "2026-07-09T17:35:00+00:00")
+    )
+    within_window = bool(
+        window_start
+        and window_end
+        and window_start <= now <= window_end
+    )
+
+    if not within_window and not closeout:
+        return {
+            "skipped": True,
+            "reason": "outside_stabilization_window",
+            "closeout": closeout,
+            "overall_pass": True,
+            "window": {
+                "start": window_start.isoformat() if window_start else None,
+                "end": window_end.isoformat() if window_end else None,
+                "within": within_window,
+            },
+            "checks": {},
+        }
+
+    baseline = params.get("baseline") or {}
+    expected_workstreams = int(baseline.get("workstreams", 640))
+    expected_tasks = int(baseline.get("tasks", 4002))
+    expected_topics = int(baseline.get("topics", 14))
+    sweep_limit = _bounded_int(params.get("sweep_limit", 6), default=6, minimum=1, maximum=24)
+    triage_max_age_hours = _bounded_int(
+        params.get("triage_max_age_hours", 36),
+        default=36,
+        minimum=1,
+        maximum=168,
+    )
+
+    health = _fetch_json("/state/health")
+    health_pass = isinstance(health, dict) and health.get("status") == "ok"
+
+    summary = _fetch_json("/state/summary")
+    totals = (summary or {}).get("totals") or {}
+    ws_total = int((totals.get("workstreams") or {}).get("total", -1))
+    task_total = int((totals.get("tasks") or {}).get("total", -1))
+    topic_total = int((totals.get("topics") or {}).get("total", -1))
+    totals_pass = (
+        ws_total == expected_workstreams
+        and task_total == expected_tasks
+        and topic_total == expected_topics
+    )
+
+    sweep_items = _fetch_json(
+        "/progress/",
+        {"event_type": "consistency_sweep_remote_all", "limit": sweep_limit},
+    )
+    sweep_rows = sweep_items if isinstance(sweep_items, list) else []
+    sweep_failures = 0
+    sweep_missing = 0
+    for item in sweep_rows:
+        detail = item.get("detail") or {}
+        if detail.get("exit_code") != 0 or detail.get("automation_error"):
+            sweep_failures += 1
+        missing = detail.get("skipped_missing") or []
+        if isinstance(missing, list) and missing:
+            sweep_missing += len(missing)
+    sweeps_pass = bool(sweep_rows) and sweep_failures == 0 and sweep_missing == 0
+
+    triage_items = _fetch_json(
+        "/progress/",
+        {"event_type": "daily_triage", "limit": 1},
+    )
+    triage_row = triage_items[0] if isinstance(triage_items, list) and triage_items else None
+    triage_at = _parse_iso_datetime(triage_row.get("created_at") if triage_row else None)
+    triage_age_hours = (
+        (now - triage_at).total_seconds() / 3600 if triage_at else None
+    )
+    triage_pass = bool(
+        triage_row
+        and triage_age_hours is not None
+        and triage_age_hours <= triage_max_age_hours
+    )
+
+    checks = {
+        "health": {"pass": health_pass, "status": health.get("status") if isinstance(health, dict) else None},
+        "totals": {
+            "pass": totals_pass,
+            "workstreams": ws_total,
+            "tasks": task_total,
+            "topics": topic_total,
+            "expected": {
+                "workstreams": expected_workstreams,
+                "tasks": expected_tasks,
+                "topics": expected_topics,
+            },
+        },
+        "sweeps": {
+            "pass": sweeps_pass,
+            "sampled": len(sweep_rows),
+            "failures": sweep_failures,
+            "missing": sweep_missing,
+        },
+        "daily_triage": {
+            "pass": triage_pass,
+            "last_at": triage_at.isoformat() if triage_at else None,
+            "age_hours": round(triage_age_hours, 2) if triage_age_hours is not None else None,
+            "max_age_hours": triage_max_age_hours,
+        },
+    }
+    overall_pass = all(section.get("pass") for section in checks.values())
+    result = {
+        "skipped": False,
+        "closeout": closeout,
+        "overall_pass": overall_pass,
+        "operator_signoff_needed": closeout and overall_pass,
+        "window": {
+            "start": window_start.isoformat() if window_start else None,
+            "end": window_end.isoformat() if window_end else None,
+            "within": within_window,
+        },
+        "checks": checks,
+        "source": str(params.get("source") or "activity-core"),
+    }
+    if closeout and not overall_pass:
+        raise RuntimeError(f"phase5 stabilization closeout failed: {result}")
+    return result
+
+
 CONTEXT_RESOLVER_REGISTRY["state-hub"] = StateHubContextResolver
 
 

src/activity_core/ops_evidence_sinks.py

@@ -33,7 +33,10 @@ def persist_ops_inventory_evidence(payload: dict[str, Any]) -> list[dict[str, An
     """
     results: list[dict[str, Any]] = []
     for source in payload.get("context_sources", []):
-        if not isinstance(source, dict) or source.get("type") != "ops-inventory":
+        if not isinstance(source, dict):
+            continue
+        source_type = source.get("type")
+        if source_type not in {"ops-inventory", "state-hub"}:
             continue
 
         params = source.get("params") or {}
@@ -43,6 +46,14 @@ def persist_ops_inventory_evidence(payload: dict[str, Any]) -> list[dict[str, An
 
         bind_key = _context_bind_key(source)
         probe_result = (payload.get("context") or {}).get(bind_key)
+        if isinstance(probe_result, dict) and probe_result.get("skipped"):
+            results.append({
+                "type": "state-hub-progress",
+                "status": "skipped",
+                "reason": probe_result.get("reason", "skipped"),
+                "context_key": bind_key,
+            })
+            continue
         if not isinstance(probe_result, dict):
             results.extend(
                 {
@@ -112,16 +123,23 @@ def _post_state_hub_progress(
             "context_key": context_key,
         }
 
-    compact = _compact_probe_result(probe_result)
+    if probe_result.get("checks") is not None:
+        compact = probe_result
+        summary = _phase5_summary_text(probe_result)
+        source_type = "state-hub"
+    else:
+        compact = _compact_probe_result(probe_result)
+        summary = _summary_text(compact.get("summary", {}))
+        source_type = "ops-inventory"
     body: dict[str, Any] = {
         "event_type": event_type,
         "author": sink.get("author", "activity-core"),
-        "summary": _summary_text(compact.get("summary", {})),
+        "summary": summary,
         "detail": {
             "activity_id": payload.get("activity_id"),
             "activity_core_run_id": run_id,
             "scheduled_for": payload.get("scheduled_for"),
-            "source_type": "ops-inventory",
+            "source_type": source_type,
             "context_key": context_key,
             "idempotency_key": idempotency_key,
             "probe": compact,
@@ -485,6 +503,21 @@ def _compact_access_path(access_path: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _phase5_summary_text(result: dict[str, Any]) -> str:
+    checks = result.get("checks") or {}
+    totals = checks.get("totals") or {}
+    sweeps = checks.get("sweeps") or {}
+    triage = checks.get("daily_triage") or {}
+    status = "pass" if result.get("overall_pass") else "fail"
+    mode = "closeout" if result.get("closeout") else "daily"
+    return (
+        f"Phase 5 stabilization {mode}: {status}; "
+        f"totals {totals.get('workstreams', '?')}/{totals.get('tasks', '?')}/"
+        f"{totals.get('topics', '?')}; sweeps sampled={sweeps.get('sampled', 0)}; "
+        f"daily_triage age_h={triage.get('age_hours', '?')}"
+    )
+
+
 def _summary_text(summary: dict[str, Any]) -> str:
     return (
         "Ops inventory probe: "

src/activity_core/report_sinks.py

@@ -11,14 +11,10 @@ from zoneinfo import ZoneInfo
 
 import httpx
 
+from activity_core.runtime_paths import custodian_repo_root, resolve_runtime_path
 from activity_core.state_hub_write import idempotency_headers
 
 _DEFAULT_STATE_HUB_URL = "http://127.0.0.1:8000"
-_THE_CUSTODIAN_ROOT = Path("/home/worsch/the-custodian")
-_FORBIDDEN_CUSTODIAN_ROOTS = (
-    _THE_CUSTODIAN_ROOT / "canon",
-    _THE_CUSTODIAN_ROOT / "workplans",
-)
 
 
 def persist_reports(payload: dict[str, Any]) -> list[dict[str, Any]]:
@@ -66,7 +62,7 @@ def _write_working_memory(
     report_entry: dict[str, Any],
     sink: dict[str, Any],
 ) -> dict[str, Any]:
-    directory = Path(sink.get("path", "")).expanduser()
+    directory = resolve_runtime_path(str(sink.get("path", "")))
     if not directory:
         raise ValueError("working-memory sink requires path")
 
@@ -257,7 +253,12 @@ def _local_date(scheduled_for: str | None, timezone_name: str) -> str:
 
 
 def _assert_allowed_output_path(path: Path) -> None:
-    for forbidden in _FORBIDDEN_CUSTODIAN_ROOTS:
+    root = custodian_repo_root()
+    forbidden_roots = (
+        root / "canon",
+        root / "workplans",
+    )
+    for forbidden in forbidden_roots:
         try:
             path.relative_to(forbidden)
         except ValueError:

src/activity_core/rules/executor.py

@@ -799,7 +799,9 @@ def _load_output_schema(schema_path: str) -> dict[str, Any] | None:
     if not schema_path:
         return None
 
-    path = Path(schema_path)
+    from activity_core.runtime_paths import resolve_runtime_path
+
+    path = resolve_runtime_path(schema_path)
     if not path.exists():
         return None
 

src/activity_core/runtime_paths.py

@@ -0,0 +1,37 @@
+"""Resolve repo-relative runtime paths for Custodian-owned assets."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+_DEFAULT_CUSTODIAN_ROOT = Path("/home/worsch/the-custodian")
+_DEFAULT_ACTIVITY_CORE_ROOT = Path("/etc/activity-core")
+
+_CUSTODIAN_SCHEME = "custodian://"
+_ACTIVITY_CORE_SCHEME = "activity-core://"
+
+
+def custodian_repo_root() -> Path:
+    raw = os.environ.get("CUSTODIAN_REPO_ROOT", "").strip()
+    return Path(raw).expanduser() if raw else _DEFAULT_CUSTODIAN_ROOT
+
+
+def activity_core_root() -> Path:
+    raw = os.environ.get("ACTIVITY_CORE_ROOT", "").strip()
+    return Path(raw).expanduser() if raw else _DEFAULT_ACTIVITY_CORE_ROOT
+
+
+def resolve_runtime_path(raw_path: str) -> Path:
+    """Map custodian:// and activity-core:// URIs to mounted runtime paths."""
+    value = str(raw_path or "").strip()
+    if not value:
+        return Path(value)
+
+    if value.startswith(_CUSTODIAN_SCHEME):
+        return (custodian_repo_root() / value.removeprefix(_CUSTODIAN_SCHEME)).resolve()
+
+    if value.startswith(_ACTIVITY_CORE_SCHEME):
+        return (activity_core_root() / value.removeprefix(_ACTIVITY_CORE_SCHEME)).resolve()
+
+    return Path(value).expanduser()