CUST-WP-0055 T05: workplan-first progress scope in sinks and resolvers

Prefer workplan_id in State Hub progress writes while dual-writing
workstream_id for wire-compat. Extend schedule_health, evidence/report
sinks, phase5 checks, k8s triage prose, and SCOPE terminology.

SCOPE.md

@@ -55,7 +55,7 @@ The two evaluation modes:
   scheduled datetime; event-type subscription via NATS; manual one-shot API
   trigger; one-shot schedule smoke tests for recurring definitions.
 - **Context resolution adapters**: repo-scoping (repository capability queries),
-  State Hub (domain/workstream state, SBOM status, daily triage digest, coding
+  State Hub (domain/workplan state, SBOM status, daily triage digest, coding
   retro read model), and ops inventory (bounded HTTP/HTTPS probes of a
   non-secret service inventory). The adapter registry is extensible for other
   sources.
@@ -218,7 +218,7 @@ workplan explicitly rehomes execution responsibility.
 One boundary nuance is now explicit: activity-core may post State Hub progress
 events as a configured report or evidence sink. That is acceptable because it
 records the result of an activity-core activation; it is not ownership of State
-Hub state, task lifecycle, or workstream planning.
+Hub state, task lifecycle, or workplan planning.
 
 The main drift risk is convenience creep: adding direct task tracking,
 project-phase state, or bespoke operational scripts because the Temporal
@@ -277,7 +277,7 @@ new one-off control paths.
 - **Coordinates with**: the state hub delegates maintenance automations to
   activity-core by publishing lifecycle events or by being resolved as context.
   activity-core may post progress events as report/evidence outputs, but it
-  does not own State Hub task/workstream state.
+  does not own State Hub task/workplan state.
 
 ---
 

k8s/railiance/21-custodian-runtime.yaml

@@ -61,13 +61,13 @@ data:
     3. `GET http://127.0.0.1:8000/messages/?to_agent=hub&unread_only=true`
     4. `GET http://127.0.0.1:8000/state/next_steps`
     5. `GET http://127.0.0.1:8000/workstreams/workplan-index`
-    6. Local workplan files for candidate workstreams that need closer inspection
+    6. Local workplan files for candidate workplans that need closer inspection
     7. Read-only `git status --short` for repos named in top recommendations
 
     Optional enrichment:
 
     - `GET /tasks/?workplan_id=<id>` for a top-ranked workplan (legacy alias: workstream_id)
-    - `GET /progress/?workstream_id=<id>&limit=5` for staleness confidence
+    - `GET /progress/?workplan_id=<id>&limit=5` for staleness confidence (legacy alias: workstream_id)
     - State Hub domain summaries through MCP when available
 
     ## Candidate Set
@@ -93,9 +93,9 @@ data:
     | Active with no recent progress | no progress in 14 days, or 7 days for high-priority items | `revisit` |
     | Large unstarted plan | `tasks_total >= 6`, `tasks_done == 0`, `tasks_in_progress == 0` | `split` or `revisit` |
     | Near complete | `tasks_done / tasks_total >= 0.75` and `tasks_todo <= 3` | `close-out` or `work-next` |
-    | Zero parsed tasks | active or blocked workstream with `tasks_total == 0` | `needs-consistency-sync` |
-    | Blocked but dependency appears closed | blocked workstream with cleared dependency or empty blocked reason | `revisit` |
-    | Open decision blocks execution | blocking decision connected to workstream/task | `needs-human` |
+    | Zero parsed tasks | active or blocked workplan with `tasks_total == 0` | `needs-consistency-sync` |
+    | Blocked but dependency appears closed | blocked workplan with cleared dependency or empty blocked reason | `revisit` |
+    | Open decision blocks execution | blocking decision connected to workplan/task | `needs-human` |
     | File-backed mismatch | missing index entry, stale `updated`, `needs_review`, or status disagreement | `needs-consistency-sync` |
     | Too broad for one session | many unrelated task clusters in one workplan | `split` |
     | Safety-sensitive | security, identity, secrets, legal, external reputation, money | `needs-human` |
@@ -120,7 +120,7 @@ data:
     - dependency-chain foundations: Railiance, identity, secrets, backup, HA
     - explicit `planning_priority: high` or low `planning_order`
     - active repo/domain goals
-    - closing a workstream that is already consuming coordination attention
+    - closing a workplan that is already consuming coordination attention
 
     ### Time Criticality
 
@@ -144,7 +144,7 @@ data:
 
     Score higher when the work unlocks:
 
-    - multiple downstream workstreams
+    - multiple downstream workplans
     - agent autonomy or fewer repeated setup steps
     - reliable cross-repo operation
     - future daily triage quality
@@ -201,7 +201,7 @@ data:
     ## Snapshot
 
     - Generated at: <timestamp>
-    - Workstreams: <active> active, <blocked> blocked, <finished> finished
+    - Workplans: <active> active, <blocked> blocked, <finished> finished
     - Tasks: <todo> todo, <in_progress> in progress, <blocked> blocked
     - Decisions: <open> open, <escalated> escalated
     - Inbox: <count> unread hub messages
@@ -211,7 +211,7 @@ data:
 
     | Rank | Action | Candidate | WSJF | Confidence | Why now |
     |------|--------|-----------|------|------------|---------|
-    | 1 | work-next | <workstream/task> | 0.0 | high | <short reason> |
+    | 1 | work-next | <workplan/task> | 0.0 | high | <short reason> |
 
     ## Loose Ends
 
@@ -240,7 +240,7 @@ data:
     When the State Hub API is reachable, append one progress event for the review:
 
     - `topic_id`: Custodian topic id when known
-    - `workstream_id`: `99993845-be6a-401d-be98-f8107014abed`
+    - `workplan_id`: `99993845-be6a-401d-be98-f8107014abed` (legacy alias: workstream_id)
     - `event_type`: `daily_triage`
     - `summary`: one sentence summarizing the top recommendation and loose-end count
     - `detail`: include top three recommendations, input health, and any human gates

src/activity_core/context_resolvers/state_hub.py

@@ -10,7 +10,7 @@ Supported queries:
   - workplan_index:   GET {STATE_HUB_URL}/workstreams/workplan-index
   - hub_inbox:        GET {STATE_HUB_URL}/messages/?to_agent=hub&unread_only=true
   - pending_decisions: GET {STATE_HUB_URL}/decisions/?status=open (topic_id/
-                      workstream_id/decision_type passed through as given)
+                      workplan_id or legacy workstream_id/decision_type passed through)
   - coding_retro:     latest /progress/ item with event_type=coding_retro
   - daily_triage_digest: curated scalar JSON digest for daily WSJF triage
   - recently_on_scope_hourly: POST {STATE_HUB_URL}/recently-on-scope/hourly
@@ -199,7 +199,9 @@ def _phase5_stabilization_check(params: dict[str, Any]) -> dict[str, Any]:
         }
 
     baseline = params.get("baseline") or {}
-    expected_workstreams = int(baseline.get("workstreams", 640))
+    expected_workplan_total = int(
+        baseline.get("workplans", 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)
@@ -215,11 +217,12 @@ def _phase5_stabilization_check(params: dict[str, Any]) -> dict[str, Any]:
 
     summary = _fetch_json("/state/summary")
     totals = (summary or {}).get("totals") or {}
-    ws_total = int((totals.get("workstreams") or {}).get("total", -1))
+    workplan_totals = totals.get("workplans") or totals.get("workstreams") or {}
+    ws_total = int(workplan_totals.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
+        ws_total == expected_workplan_total
         and task_total == expected_tasks
         and topic_total == expected_topics
     )
@@ -259,11 +262,13 @@ def _phase5_stabilization_check(params: dict[str, Any]) -> dict[str, Any]:
         "health": {"pass": health_pass, "status": health.get("status") if isinstance(health, dict) else None},
         "totals": {
             "pass": totals_pass,
+            "workplans": ws_total,
             "workstreams": ws_total,
             "tasks": task_total,
             "topics": topic_total,
             "expected": {
-                "workstreams": expected_workstreams,
+                "workplans": expected_workplan_total,
+                "workstreams": expected_workplan_total,
                 "tasks": expected_tasks,
                 "topics": expected_topics,
             },

src/activity_core/ops_evidence_sinks.py

@@ -10,7 +10,7 @@ from typing import Any
 import httpx
 
 from activity_core.context_resolvers.ops_inventory import _sanitize_url
-from activity_core.state_hub_write import idempotency_headers
+from activity_core.state_hub_write import apply_progress_scope_fields, idempotency_headers
 
 _DEFAULT_STATE_HUB_URL = "http://127.0.0.1:8000"
 _INTER_HUB_SINK_TYPES = {
@@ -149,9 +149,7 @@ def _post_state_hub_progress(
             "probe": compact,
         },
     }
-    for key in ("topic_id", "workstream_id", "task_id", "decision_id"):
-        if sink.get(key):
-            body[key] = sink[key]
+    apply_progress_scope_fields(body, sink)
 
     resp = httpx.post(
         f"{base_url}/progress/",
@@ -529,7 +527,8 @@ def _phase5_summary_text(result: dict[str, Any]) -> str:
     mode = "closeout" if result.get("closeout") else "daily"
     return (
         f"Phase 5 stabilization {mode}: {status}; "
-        f"totals {totals.get('workstreams', '?')}/{totals.get('tasks', '?')}/"
+        f"workplan totals {totals.get('workplans', totals.get('workstreams', '?'))}/"
+        f"{totals.get('tasks', '?')}/"
         f"{totals.get('topics', '?')}; sweeps sampled={sweeps.get('sampled', 0)}; "
         f"daily_triage age_h={triage.get('age_hours', '?')}"
     )

src/activity_core/report_sinks.py

@@ -16,7 +16,7 @@ from activity_core.runtime_paths import (
     custodian_repo_root,
     resolve_runtime_path,
 )
-from activity_core.state_hub_write import idempotency_headers
+from activity_core.state_hub_write import apply_progress_scope_fields, idempotency_headers
 
 _DEFAULT_STATE_HUB_URL = "http://127.0.0.1:8000"
 
@@ -146,9 +146,7 @@ def _post_state_hub_progress(
         body["detail"]["working_memory_status"] = report_entry.get(
             "working_memory_status"
         )
-    for key in ("topic_id", "workstream_id", "task_id", "decision_id"):
-        if sink.get(key):
-            body[key] = sink[key]
+    apply_progress_scope_fields(body, sink)
 
     resp = httpx.post(
         f"{base_url}/progress/",

src/activity_core/schedule_health.py

@@ -139,6 +139,7 @@ def post_missed_fire_alert(
     state_hub_url: str | None = None,
     author: str = "activity-core",
     topic_id: str | None = None,
+    workplan_id: str | None = None,
     workstream_id: str | None = None,
     timeout_seconds: float = 10.0,
 ) -> dict[str, Any]:
@@ -174,8 +175,10 @@ def post_missed_fire_alert(
     }
     if topic_id:
         body["topic_id"] = topic_id
-    if workstream_id:
-        body["workstream_id"] = workstream_id
+    scope_id = workplan_id or workstream_id
+    if scope_id:
+        body["workplan_id"] = scope_id
+        body["workstream_id"] = scope_id
 
     # Dedup repeated alerts for the same missed window (same schedule + last fire).
     last_fired = health.last_fired_at.isoformat() if health.last_fired_at else "none"

src/activity_core/state_hub_write.py

@@ -15,9 +15,26 @@ ignored by State Hub versions that do not yet honour it.
 
 from __future__ import annotations
 
+from typing import Any
+
 IDEMPOTENCY_HEADER = "Idempotency-Key"
 
 
+def apply_progress_scope_fields(body: dict[str, Any], source: dict[str, Any]) -> None:
+    """Copy topic/workplan/task/decision scope from a sink config into a progress body.
+
+    Prefers ``workplan_id``; dual-writes ``workstream_id`` for wire-compat until
+    legacy-meter retires the alias.
+    """
+    workplan_id = source.get("workplan_id") or source.get("workstream_id")
+    if workplan_id:
+        body["workplan_id"] = workplan_id
+        body["workstream_id"] = workplan_id
+    for key in ("topic_id", "task_id", "decision_id"):
+        if source.get(key):
+            body[key] = source[key]
+
+
 def idempotency_key(*parts: str | None) -> str:
     """Build a stable, header-safe idempotency key from identity parts.
 

tests/test_ops_evidence_sinks.py

@@ -123,6 +123,7 @@ def test_state_hub_progress_sink_posts_compact_probe_summary(monkeypatch) -> Non
     ]
     body = posts[0]["json"]
     assert body["summary"] == "Ops inventory probe: 1 ok, 0 degraded, 0 down, 1 skipped"
+    assert body["workplan_id"] == "workstream-1"
     assert body["workstream_id"] == "workstream-1"
     assert body["task_id"] == "task-1"
     assert body["detail"]["activity_core_run_id"] == _run_id()
@@ -136,6 +137,35 @@ def test_state_hub_progress_sink_posts_compact_probe_summary(monkeypatch) -> Non
     assert "token=secret" not in serialized
 
 
+def test_state_hub_progress_prefers_workplan_id(monkeypatch) -> None:
+    posts: list[dict[str, Any]] = []
+
+    def fake_get(url: str, **kwargs: Any) -> DummyResponse:
+        return DummyResponse([])
+
+    def fake_post(url: str, **kwargs: Any) -> DummyResponse:
+        posts.append({"url": url, **kwargs})
+        return DummyResponse({"id": "progress-2"})
+
+    monkeypatch.setattr(httpx, "get", fake_get)
+    monkeypatch.setattr(httpx, "post", fake_post)
+
+    persist_ops_inventory_evidence(
+        _payload([
+            {
+                "type": "state-hub-progress",
+                "state_hub_url": "http://state-hub.test",
+                "event_type": "ops_inventory_probe",
+                "workplan_id": "workplan-1",
+            }
+        ])
+    )
+
+    body = posts[0]["json"]
+    assert body["workplan_id"] == "workplan-1"
+    assert body["workstream_id"] == "workplan-1"
+
+
 def test_core_hub_stabilization_sink_posts_progress(monkeypatch) -> None:
     posts: list[dict[str, Any]] = []
 

tests/test_report_sinks.py

@@ -116,6 +116,7 @@ def test_state_hub_progress_sink_posts(monkeypatch) -> None:
         }
     ]
     assert posts[0]["url"] == "http://state-hub.test/progress/"
+    assert posts[0]["json"]["workplan_id"] == "workstream-1"
     assert posts[0]["json"]["workstream_id"] == "workstream-1"
     assert posts[0]["json"]["detail"]["activity_core_run_id"] == payload_run_id()
     assert posts[0]["json"]["detail"]["output_validated"] is True

tests/test_schedule_health.py

@@ -4,7 +4,9 @@ from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
 
-from activity_core.schedule_health import evaluate_schedule_health
+import httpx
+
+from activity_core.schedule_health import evaluate_schedule_health, post_missed_fire_alert
 
 NOW = datetime(2026, 6, 23, 12, 0, tzinfo=timezone.utc)
 
@@ -70,6 +72,35 @@ def test_no_fire_recorded_for_due_schedule_is_unhealthy() -> None:
     assert "no recorded fire" in health.reasons[0]
 
 
+def test_post_missed_fire_alert_dual_writes_workplan_scope(monkeypatch) -> None:
+    posts: list[dict] = []
+
+    class _Resp:
+        def raise_for_status(self) -> None: ...
+        def json(self) -> dict[str, str]:
+            return {"id": "progress-1"}
+
+    monkeypatch.setattr(httpx, "post", lambda url, **kwargs: posts.append(kwargs) or _Resp())
+
+    health = evaluate_schedule_health(
+        activity_id="daily",
+        missed_catchup_window=1,
+        last_fired_at=NOW - timedelta(days=2),
+        now=NOW,
+        expected_interval=timedelta(days=1),
+    )
+    result = post_missed_fire_alert(
+        health,
+        state_hub_url="http://state-hub.test",
+        workplan_id="wp-123",
+    )
+
+    assert result["status"] == "posted"
+    body = posts[0]["json"]
+    assert body["workplan_id"] == "wp-123"
+    assert body["workstream_id"] == "wp-123"
+
+
 def test_no_interval_and_no_fire_is_not_flagged() -> None:
     # Without an expected interval we cannot assert a miss from absence alone.
     health = evaluate_schedule_health(

tests/test_state_hub_context_resolver.py

@@ -751,3 +751,5 @@ def test_phase5_stabilization_check_passes(monkeypatch) -> None:
     assert result["overall_pass"] is True
     assert result["skipped"] is False
     assert result["checks"]["totals"]["pass"] is True
+    assert result["checks"]["totals"]["workplans"] == 640
+    assert result["checks"]["totals"]["workstreams"] == 640

tests/test_state_hub_write.py

@@ -8,6 +8,7 @@ import pytest
 from activity_core import report_sinks
 from activity_core.state_hub_write import (
     IDEMPOTENCY_HEADER,
+    apply_progress_scope_fields,
     idempotency_headers,
     idempotency_key,
 )
@@ -33,6 +34,24 @@ def test_headers_carry_the_key() -> None:
     assert headers == {IDEMPOTENCY_HEADER: "run1:i:e"}
 
 
+def test_apply_progress_scope_prefers_workplan_id() -> None:
+    body: dict[str, str] = {}
+    apply_progress_scope_fields(
+        body,
+        {"workplan_id": "wp-1", "workstream_id": "ws-legacy", "topic_id": "topic-1"},
+    )
+    assert body["workplan_id"] == "wp-1"
+    assert body["workstream_id"] == "wp-1"
+    assert body["topic_id"] == "topic-1"
+
+
+def test_apply_progress_scope_dual_writes_legacy_workstream_id() -> None:
+    body: dict[str, str] = {}
+    apply_progress_scope_fields(body, {"workstream_id": "ws-1"})
+    assert body["workplan_id"] == "ws-1"
+    assert body["workstream_id"] == "ws-1"
+
+
 def test_distinct_identities_get_distinct_keys() -> None:
     assert idempotency_key("r", "i", "daily_triage") != idempotency_key(
         "r", "i", "schedule_miss"