diff --git a/k8s/railiance/21-custodian-runtime.yaml b/k8s/railiance/21-custodian-runtime.yaml index 8d73ed0..a73dec6 100644 --- a/k8s/railiance/21-custodian-runtime.yaml +++ b/k8s/railiance/21-custodian-runtime.yaml @@ -66,7 +66,7 @@ data: Optional enrichment: - - `GET /tasks/?workstream_id=` for a top-ranked workstream + - `GET /tasks/?workplan_id=` for a top-ranked workplan (legacy alias: workstream_id) - `GET /progress/?workstream_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 diff --git a/src/activity_core/context_resolvers/state_hub.py b/src/activity_core/context_resolvers/state_hub.py index a598c25..b899758 100644 --- a/src/activity_core/context_resolvers/state_hub.py +++ b/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", []): - if workstream.get("status") not in _OPEN_WORKSTREAM_STATUSES: + topic_workplans = topic.get("workplans") or topic.get("workstreams") or [] + for workplan_row in topic_workplans: + if workplan_row.get("status") not in _OPEN_WORKPLAN_STATUSES: continue - workstream_id = workstream.get("id") - detail = _fetch_json(f"/workstreams/{workstream_id}") if workstream_id else {} - tasks = _fetch_json("/tasks/", {"workstream_id": workstream_id, "limit": 200}) + workplan_id = workplan_row.get("id") + detail = _fetch_json(f"/workstreams/{workplan_id}") if workplan_id else {} + 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, - "slug": workstream.get("slug"), - "title": _short_text(workstream.get("title", ""), 120), + "id": workplan_id, + "slug": workplan_row.get("slug"), + "title": _short_text(workplan_row.get("title", ""), 120), "domain": domain, "repo_slug": indexed.get("repo_slug"), - "status": workstream.get("status"), - "owner": workstream.get("owner"), + "status": workplan_row.get("status"), + "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"), - "workstream_slug": item.get("workstream_slug"), - "workstream_title": _short_text(item.get("workstream_title", ""), 120), + "workplan_id": workplan_id, + "workplan_slug": workplan_slug, + "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), } diff --git a/tests/test_state_hub_context_resolver.py b/tests/test_state_hub_context_resolver.py index a15a5ae..951fd0d 100644 --- a/tests/test_state_hub_context_resolver.py +++ b/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"] == {