STATE-WP-0069 T06: add workplan flow entity and prefer open_workplans

Introduce flows/workplan.yaml and /flows/workplan/* routes alongside the
legacy workstream flow. Dashboard, README, and MCP flow docs now prefer
open_workplans with open_workstreams fallback until dual-key retirement.

README.md

@@ -186,7 +186,8 @@ Returns a full snapshot in one call — used by both the MCP server and dashboar
   "blocking_decisions": [...], // pending decisions only
   "waiting_tasks": [...],
   "recent_progress": [...],    // last 20 events
-  "open_workstreams": [...]
+  "open_workplans": [...],
+  "open_workstreams": [...]   // legacy alias of open_workplans until STATE-WP-0069 T06 retires it
 }
 ```
 

api/routers/flows.py

@@ -24,12 +24,19 @@ from api.workplan_status import normalize_workplan_status
 
 router = APIRouter(prefix="/flows", tags=["flows"])
 
+_WORKPLAN_FLOW_ENTITY_TYPES = frozenset({"workplan", "workstream"})
+
+
+def _is_workplan_flow(entity_type: str) -> bool:
+    return entity_type in _WORKPLAN_FLOW_ENTITY_TYPES
+
 
 @router.get("/definitions")
 async def list_flow_definitions() -> list[dict[str, Any]]:
     flows = [
         load_flow(entity_type)
         for entity_type in (
+            "workplan",
             "workstream",
             "task",
             "contribution",
@@ -93,7 +100,7 @@ async def advance_workstation(
         )
 
     entity = await _entity(entity_type, entity_id, session)
-    if entity_type == "workstream":
+    if _is_workplan_flow(entity_type):
         transition_workplan_status(entity, target_workstation)
     elif entity_type == "task":
         parent = await session.get(Workplan, entity.workplan_id)
@@ -117,7 +124,9 @@ async def _flow_object(
 ) -> dict[str, Any]:
     entity = await _entity(entity_type, entity_id, session)
     status = _value(entity.status)
-    current_status = normalize_workplan_status(status) if entity_type == "workstream" else status
+    current_status = (
+        normalize_workplan_status(status) if _is_workplan_flow(entity_type) else status
+    )
     obj: dict[str, Any] = {
         "id": str(entity.id),
         "status": current_status,
@@ -125,7 +134,7 @@ async def _flow_object(
         "previous_workstation": current_status,
     }
 
-    if entity_type == "workstream":
+    if _is_workplan_flow(entity_type):
         tasks = list((await session.execute(
             select(Task).where(Task.workplan_id == entity_id)
         )).scalars().all())
@@ -163,6 +172,7 @@ async def _entity(
     session: AsyncSession,
 ):
     model_by_type = {
+        "workplan": Workplan,
         "workstream": Workplan,
         "task": Task,
         "contribution": Contribution,

dashboard/src/data/summary.json.py

@@ -37,5 +37,6 @@ except urllib.error.URLError as e:
         "waiting_tasks": [],
         "blocked_tasks": [],
         "recent_progress": [],
+        "open_workplans": [],
         "open_workstreams": [],
     }))

dashboard/src/docs/dependencies.md

@@ -86,5 +86,5 @@ a cycle exists and the WHI is penalised by 50%. The WHI KPI card on the
 
 ## Data source
 
-Dependency edges are derived from the `depends_on` arrays on `open_workstreams`
+Dependency edges are derived from the `depends_on` arrays on `open_workplans`
 in `GET /state/summary`. Polls every **15 seconds**.

dashboard/src/index.md

@@ -437,7 +437,7 @@ display(html`<div class="grid grid-cols-3" style="gap:1rem;margin-bottom:1.5rem"
 
 ```js
 const waitingTasks = summary.waiting_tasks ?? summary.blocked_tasks ?? [];
-const wsById = Object.fromEntries((summary.open_workstreams ?? []).map(w => [w.id, w]));
+const wsById = Object.fromEntries((summary.open_workplans ?? summary.open_workstreams ?? []).map(w => [w.id, w]));
 const todayCount = (summary.recent_progress ?? []).filter(e =>
   e.created_at?.startsWith(new Date().toISOString().slice(0, 10))).length;
 const decCount = (decisions.open ?? 0) + (decisions.escalated ?? 0);

dashboard/src/workstreams.md

@@ -282,7 +282,7 @@ display(Plot.plot({
 display(_filtersForm);
 
 {
-  // Enrich each workplan with tasks/deps data from open_workstreams summary
+  // Enrich each workplan with tasks/deps data from open_workplans summary
   const _openWsMap = Object.fromEntries(openWs.map(w => [w.id, w]));
   const _wsTable = buildEntityTable(
     filtered,
@@ -304,7 +304,7 @@ display(_filtersForm);
 ## Dependencies
 
 ```js
-// Build dep cards from the enriched open_workstreams in the summary
+// Build dep cards from the enriched open_workplans in the summary
 const wsWithDeps = openWs.filter(w => {
   const domain = data.find(d => d.id === w.id)?.domain ?? "unknown";
   return (filters.domain.length === 0 || filters.domain.includes(domain)) &&

docs/workplan-terminology-legacy-retirement-backlog.md

@@ -61,7 +61,7 @@ Risk order: REST > MCP > events > dashboard prose > internal identifiers.
 | 5 | Dashboard nav label `Workstreams` | `Workplans` (URL compat retained) | Prose | T02 |
 | 5 | `dashboard/src/index.md` user-facing copy | workplan-first strings | Prose | T02 |
 | 6 | `open_workstreams` summary cache key | `open_workplans` | Internal | T06 |
-| 6 | `flows/workstream.yaml` entity id | workplan successor flow | Internal | T06 |
+| 6 | `flows/workstream.yaml` entity id | `flows/workplan.yaml` (`custodian.workplan.v1`) | Internal | T06 (workplan flow shipped; workstream yaml retained) |
 
 ### Query-param aliases (not separately metered today)
 

flows/workplan.yaml

@@ -0,0 +1,59 @@
+id: custodian.workplan.v1
+entity_type: workplan
+workstations:
+  - name: proposed
+    description: Plan exists but needs review against the current repo state.
+    entry_assertions: []
+    exit_assertions: []
+  - name: ready
+    description: Reviewed and ready to execute.
+    entry_assertions: []
+    exit_assertions: []
+  - name: active
+    description: Work is underway.
+    entry_assertions: []
+    exit_assertions:
+      - id: dependencies.all_complete
+        target: dependencies.*.workstation
+        op: all_eq
+        value:
+          - finished
+          - archived
+        description: Dependency workplans have reached a closed state.
+  - name: blocked
+    description: Work is blocked by incomplete dependencies or missing input.
+    entry_assertions:
+      - id: dependencies.any_incomplete
+        target: dependencies.*.workstation
+        op: custom
+        value:
+          - finished
+          - archived
+        description: At least one dependency is not finished or archived.
+    exit_assertions:
+      - id: dependencies.all_complete
+        target: dependencies.*.workstation
+        op: all_eq
+        value:
+          - finished
+          - archived
+        description: All dependency workplans have reached finished or archived.
+  - name: backlog
+    description: Intentionally parked for later.
+    entry_assertions: []
+    exit_assertions: []
+  - name: finished
+    description: Work is complete.
+    entry_assertions:
+      - id: tasks.all_done
+        target: tasks.*.status
+        op: all_eq
+        value:
+          - done
+          - cancel
+        description: All child tasks are done or canceled.
+    exit_assertions: []
+  - name: archived
+    description: Closed work has been moved out of the active set.
+    entry_assertions: []
+    exit_assertions: []

mcp_server/server.py

@@ -577,7 +577,7 @@ def get_flow_state(entity_type: str, entity_id: str) -> str:
     """Return the declarative flow state for one entity.
 
     Args:
-        entity_type: workstream | task | contribution | capability_request
+        entity_type: workplan | workstream | task | contribution | capability_request
         entity_id: UUID of the entity
 
     Returns current workstation, exit-blocking assertions, reachable
@@ -592,7 +592,7 @@ def advance_workstation(entity_type: str, entity_id: str, target_workstation: st
     """Attempt to move an entity to a target workstation.
 
     Args:
-        entity_type: workstream | task | contribution | capability_request
+        entity_type: workplan | workstream | task | contribution | capability_request
         entity_id: UUID of the entity
         target_workstation: desired workstation/status name
 

tests/test_routers_core.py

@@ -507,7 +507,22 @@ class TestFlowEndpoints:
         r = await client.get("/flows/definitions")
         assert r.status_code == 200
         entity_types = {item["entity_type"] for item in r.json()}
-        assert {"workstream", "task", "contribution", "capability_request"} <= entity_types
+        assert {"workplan", "workstream", "task", "contribution", "capability_request"} <= entity_types
+
+    async def test_get_flow_state_and_advance_workplan(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        ws = await _create_workstream(client, topic["id"])
+        task = await _create_task(client, ws["id"])
+        await client.patch(f"/tasks/{task['id']}", json={"status": "done"})
+
+        r = await client.get(f"/flows/workplan/{ws['id']}")
+        assert r.status_code == 200
+        assert r.json()["current_workstation"] == "active"
+
+        r = await client.post(f"/flows/workplan/{ws['id']}/advance/finished")
+        assert r.status_code == 200
+        assert r.json()["current_workstation"] == "finished"
 
     async def test_get_flow_state_and_advance_workstream(self, client):
         await _create_domain(client)

tests/test_task_flow_engine.py

@@ -173,9 +173,27 @@ def test_yaml_flow_definitions_load_and_evaluate_representative_entities():
         "capability_request",
         "contribution",
         "task",
+        "workplan",
         "workstream",
     ]
 
+    workplan_result = FlowEngine(
+        custom_ops={
+            "dependencies.any_incomplete": lambda assertion, obj, values: any(
+                value not in assertion.value for value in values
+            )
+        }
+    ).evaluate(
+        {
+            "status": "active",
+            "tasks": [{"status": "done"}],
+            "dependencies": [{"workstation": "finished"}],
+        },
+        flows["workplan"],
+    )
+    assert "finished" in workplan_result.reachable
+    assert "blocked" in [item.workstation for item in workplan_result.unreachable]
+
     workstream_result = FlowEngine(
         custom_ops={
             "dependencies.any_incomplete": lambda assertion, obj, values: any(

workplans/STATE-WP-0069-workplan-terminology-legacy-retirement.md

@@ -10,7 +10,7 @@ topic_slug: custodian
 planning_priority: medium
 planning_order: 69
 created: "2026-07-08"
-updated: "2026-07-08"
+updated: "2026-07-09"
 state_hub_workstream_id: "923bb94a-d16c-422c-b81e-16328bd7b60c"
 ---
 
@@ -213,6 +213,11 @@ Done when internal code reads workplan-first and grep budget for
 Progress 2026-07-08: `/state/summary` now dual-writes `open_workplans` alongside
 legacy `open_workstreams`; MCP `get_domain_summary` prefers `open_workplans`.
 
+Progress 2026-07-09 (T06): added `flows/workplan.yaml` (`custodian.workplan.v1`) and
+`/flows/workplan/{id}` advance path; dashboard and README now read `open_workplans`
+first with `open_workstreams` fallback. Legacy `flows/workstream.yaml` and summary
+dual-key remain until legacy-meter clears callers.
+
 ## Task: Closeout — registry cleanup and verification
 
 ```task