refactor(terminology): workplan-first state hub resolver internals (CUST-WP-0055 T05)

Rename open-workplan digest helpers, expose open_workplans alongside legacy
open_workstreams in daily triage digests, and update custodian runtime prompt
prose while preserving wire-compat API paths and query params.

k8s/railiance/21-custodian-runtime.yaml

@@ -66,7 +66,7 @@ data:
 
     Optional enrichment:
 
-    - `GET /tasks/?workstream_id=<id>` for a top-ranked workstream
+    - `GET /tasks/?workplan_id=<id>` for a top-ranked workplan (legacy alias: workstream_id)
     - `GET /progress/?workstream_id=<id>&limit=5` for staleness confidence
     - State Hub domain summaries through MCP when available
 
@@ -74,11 +74,11 @@ data:
 
     Build the candidate list from:
 
-    - all open workstreams in `state_summary.open_workstreams`
+    - all open workplans in `state_summary.open_workstreams` (legacy summary key)
     - all derived `state_summary.next_steps`
     - blocked tasks and blocking decisions from the summary
     - high-priority file-backed workplans surfaced by `workplan-index`
-    - workstreams with suspicious structure, such as zero parsed tasks or stale
+    - workplans with suspicious structure, such as zero parsed tasks or stale
       active plans
 
     Keep the scored table compact. Score at most 15 candidates internally and

src/activity_core/context_resolvers/state_hub.py

@@ -36,7 +36,7 @@ from activity_core.context_resolvers.base import CONTEXT_RESOLVER_REGISTRY, Cont
 _DEFAULT_STATE_HUB_URL = "http://127.0.0.1:8000"
 _TIMEOUT_SECONDS = 10.0
 _SWEEP_TIMEOUT_SECONDS = 330.0
-_OPEN_WORKSTREAM_STATUSES = {"active", "ready", "blocked"}
+_OPEN_WORKPLAN_STATUSES = {"active", "ready", "blocked"}
 _OPEN_TASK_STATUSES = {"wait", "todo", "progress"}
 # Sentinel age for repos that have never had an SBOM ingested. Large enough
 # that any threshold-based staleness rule treats them as "very stale" without
@@ -693,9 +693,9 @@ def _daily_triage_digest(params: dict[str, Any]) -> str:
     if not isinstance(inbox, list):
         inbox = []
 
-    max_workstreams = int(params.get("max_workstreams", 12))
+    max_workplans = int(params.get("max_workplans", params.get("max_workstreams", 12)))
     max_next_steps = int(params.get("max_next_steps", 8))
-    open_workstreams = _open_workstream_digest(summary, workplan_index, max_workstreams)
+    open_workplans = _open_workplan_digest(summary, workplan_index, max_workplans)
     ranked_suggestions = summary.get("ranked_suggestions") or []
     if not isinstance(ranked_suggestions, list):
         ranked_suggestions = []
@@ -703,7 +703,8 @@ def _daily_triage_digest(params: dict[str, Any]) -> str:
     digest = {
         "generated_at": summary.get("generated_at"),
         "totals": summary.get("totals", {}),
-        "open_workstreams": open_workstreams,
+        "open_workplans": open_workplans,
+        "open_workstreams": open_workplans,
         "next_steps": [_safe_next_step(item) for item in next_steps[:max_next_steps]],
         "ranked_suggestions": [
             _safe_ranked_suggestion(item)
@@ -729,35 +730,36 @@ def _daily_triage_digest(params: dict[str, Any]) -> str:
     return json.dumps(digest, sort_keys=True, separators=(",", ":"))
 
 
-def _open_workstream_digest(
+def _open_workplan_digest(
     summary: dict[str, Any],
     workplan_index: dict[str, Any],
-    max_workstreams: int,
+    max_workplans: int,
 ) -> list[dict[str, Any]]:
-    index = workplan_index.get("workstreams") or {}
+    index = workplan_index.get("workplans") or workplan_index.get("workstreams") or {}
     candidates: list[dict[str, Any]] = []
     for topic in summary.get("topics", []):
         domain = topic.get("domain_slug") or topic.get("slug")
-        for workstream in topic.get("workstreams", []):
+        topic_workplans = topic.get("workplans") or topic.get("workstreams") or []
-            if workstream.get("status") not in _OPEN_WORKSTREAM_STATUSES:
+        for workplan_row in topic_workplans:
+            if workplan_row.get("status") not in _OPEN_WORKPLAN_STATUSES:
                 continue
-            workstream_id = workstream.get("id")
+            workplan_id = workplan_row.get("id")
-            detail = _fetch_json(f"/workstreams/{workstream_id}") if workstream_id else {}
+            detail = _fetch_json(f"/workstreams/{workplan_id}") if workplan_id else {}
-            tasks = _fetch_json("/tasks/", {"workstream_id": workstream_id, "limit": 200})
+            tasks = _fetch_json("/tasks/", {"workstream_id": workplan_id, "limit": 200})
             if not isinstance(detail, dict):
                 detail = {}
             if not isinstance(tasks, list):
                 tasks = []
             counts = _task_counts(tasks)
-            indexed = index.get(workstream_id, {}) if isinstance(index, dict) else {}
+            indexed = index.get(workplan_id, {}) if isinstance(index, dict) else {}
             candidates.append({
-                "id": workstream_id,
+                "id": workplan_id,
-                "slug": workstream.get("slug"),
+                "slug": workplan_row.get("slug"),
-                "title": _short_text(workstream.get("title", ""), 120),
+                "title": _short_text(workplan_row.get("title", ""), 120),
                 "domain": domain,
                 "repo_slug": indexed.get("repo_slug"),
-                "status": workstream.get("status"),
+                "status": workplan_row.get("status"),
-                "owner": workstream.get("owner"),
+                "owner": workplan_row.get("owner"),
                 "planning_priority": detail.get("planning_priority"),
                 "planning_order": detail.get("planning_order"),
                 "file": indexed.get("relative_path"),
@@ -768,7 +770,10 @@ def _open_workstream_digest(
             })
 
     candidates.sort(key=_candidate_sort_key)
-    return candidates[:max_workstreams]
+    return candidates[:max_workplans]
+
+
+_open_workstream_digest = _open_workplan_digest
 
 
 def _task_counts(tasks: list[dict[str, Any]]) -> dict[str, int]:
@@ -799,12 +804,18 @@ def _representative_tasks(tasks: list[dict[str, Any]], limit: int) -> list[dict[
 
 
 def _safe_next_step(item: dict[str, Any]) -> dict[str, Any]:
+    workplan_id = item.get("workplan_id") or item.get("workstream_id")
+    workplan_slug = item.get("workplan_slug") or item.get("workstream_slug")
+    workplan_title = item.get("workplan_title") or item.get("workstream_title")
     return {
         "type": item.get("type"),
         "domain": item.get("domain"),
-        "workstream_id": item.get("workstream_id"),
+        "workplan_id": workplan_id,
-        "workstream_slug": item.get("workstream_slug"),
+        "workplan_slug": workplan_slug,
-        "workstream_title": _short_text(item.get("workstream_title", ""), 120),
+        "workplan_title": _short_text(workplan_title, 120),
+        "workstream_id": workplan_id,
+        "workstream_slug": workplan_slug,
+        "workstream_title": _short_text(workplan_title, 120),
         "task_id": item.get("task_id"),
         "task_title": _short_text(item.get("task_title", ""), 120),
     }

tests/test_state_hub_context_resolver.py

@@ -690,6 +690,7 @@ def test_daily_triage_digest_is_curated_scalar_json(monkeypatch) -> None:
     import json
     digest = json.loads(raw_digest)
     assert digest["totals"] == {"tasks": {"todo": 4, "wait": 1}}
+    assert digest["open_workplans"] == digest["open_workstreams"]
     assert digest["open_workstreams"][0]["slug"] == "cust-wp-0045"
     assert digest["open_workstreams"][0]["planning_priority"] == "high"
     assert digest["open_workstreams"][0]["open_task_counts"] == {