Close lifecycle transition helper workplan

api/routers/flows.py

@@ -16,9 +16,10 @@ from api.flow_defs import (
 )
 from api.models.capability_request import CapabilityRequest
 from api.models.contribution import Contribution
-from api.models.task import Task
+from api.models.task import Task, TaskStatus
 from api.models.workstream import Workstream
 from api.models.workstream_dependency import WorkstreamDependency
+from api.services.lifecycle import transition_task_status, transition_workstream_status
 from api.workplan_status import normalize_workstream_status
 
 router = APIRouter(prefix="/flows", tags=["flows"])
@@ -92,7 +93,18 @@ async def advance_workstation(
         )
 
     entity = await _entity(entity_type, entity_id, session)
-    entity.status = target_workstation
+    if entity_type == "workstream":
+        transition_workstream_status(entity, target_workstation)
+    elif entity_type == "task":
+        parent = await session.get(Workstream, entity.workstream_id)
+        transition_task_status(
+            entity,
+            target_workstation,
+            parent_workstream=parent,
+            status_coercer=TaskStatus,
+        )
+    else:
+        entity.status = target_workstation
     await session.commit()
     await session.refresh(entity)
     return await get_flow_state(entity_type, entity_id, session)

api/routers/reconciliation.py

@@ -11,7 +11,12 @@ from api.models.task import Task
 from api.models.task import TaskStatus
 from api.models.workstream import Workstream
 from api.schemas.reconciliation import StateChangeRequest, StateChangeResponse
-from api.services.lifecycle import status_value
+from api.services.lifecycle import (
+    should_activate_parent_for_task_start,
+    status_value,
+    transition_task_status,
+    transition_workstream_status,
+)
 from api.services.reconciliation import (
     ReconciliationClass,
     StateChangeClassification,
@@ -172,7 +177,7 @@ async def classify_state_change(
                             )
                             conflict = True
                         else:
-                            ws.status = target_status
+                            transition_workstream_status(ws, target_status)
                             await session.commit()
                             write_result = "applied"
 
@@ -281,24 +286,54 @@ async def classify_state_change(
                 )
                 conflict = True
             else:
+                original_text = None
+                parent_will_activate = should_activate_parent_for_task_start(
+                    previous_task_status=current_status,
+                    new_task_status=target_status,
+                    parent_workstream_status=ws.status if ws else None,
+                )
                 try:
+                    original_text = workplan_ref.path.read_text(encoding="utf-8")
                     patch_task_status(workplan_ref.path, task.id, target_status)
                     patched_status = status_value(task_block_status(workplan_ref.path, task.id))
+                    if parent_will_activate:
+                        patch_workplan_status(workplan_ref.path, "active")
+                        parent_status = normalize_workstream_status(workplan_status(workplan_ref.path))
+                        if parent_status != "active":
+                            if original_text is not None:
+                                workplan_ref.path.write_text(original_text, encoding="utf-8")
+                            classification = _conflict(
+                                "parent workplan status could not be patched to 'active'",
+                                "inspect the workplan frontmatter format before retrying",
+                            )
+                            conflict = True
                 except OSError as exc:
+                    if original_text is not None:
+                        workplan_ref.path.write_text(original_text, encoding="utf-8")
                     classification = _conflict(
                         f"workplan task write failed: {exc}",
                         "fix repo file access and retry the reconciliation",
                     )
                     conflict = True
                 else:
-                    if patched_status != target_status:
+                    if conflict:
+                        pass
+                    elif patched_status != target_status:
+                        if original_text is not None:
+                            workplan_ref.path.write_text(original_text, encoding="utf-8")
                         classification = _conflict(
                             f"workplan task block could not be patched to {target_status!r}",
                             "inspect the task block format before retrying",
                         )
                         conflict = True
                     else:
-                        task.status = TaskStatus(target_status)
+                        transition = transition_task_status(
+                            task,
+                            target_status,
+                            parent_workstream=ws,
+                            previous_task_status=current_status,
+                            status_coercer=TaskStatus,
+                        )
                         if body.blocking_reason is not None:
                             task.blocking_reason = body.blocking_reason
                         await session.commit()

api/routers/tasks.py

@@ -10,7 +10,7 @@ from api.models.task import Task, TaskStatus
 from api.models.token_event import TokenEvent
 from api.models.workstream import Workstream
 from api.schemas.task import TaskCreate, TaskRead, TaskUpdate
-from api.services.lifecycle import activate_parent_for_task_start, status_value
+from api.services.lifecycle import status_value, transition_task_status
 
 router = APIRouter(prefix="/tasks", tags=["tasks"])
 
@@ -52,10 +52,11 @@ async def create_task(
     session.add(task)
     if status_value(task.status) == "in_progress":
         ws = await session.get(Workstream, task.workstream_id)
-        activate_parent_for_task_start(
+        transition_task_status(
-            previous_task_status="todo",
+            task,
-            new_task_status=task.status,
+            task.status,
             parent_workstream=ws,
+            previous_task_status="todo",
         )
     await session.commit()
     await session.refresh(task)
@@ -100,17 +101,18 @@ async def update_task(
     update_data = body.model_dump(exclude_unset=True)
     token_data = {k: update_data.pop(k) for k in list(update_data.keys()) if k in token_field_names}
     suppress_token_event = bool(token_data.pop("suppress_token_event", False))
-    status_update = update_data.get("status")
+    status_update = update_data.pop("status", None)
     new_status = status_value(status_update) if status_update is not None else None
 
     for field, value in update_data.items():
         setattr(task, field, value)
     if new_status is not None:
         ws = await session.get(Workstream, task.workstream_id)
-        activate_parent_for_task_start(
+        transition_task_status(
-            previous_task_status=previous_status,
+            task,
-            new_task_status=new_status,
+            status_update,
             parent_workstream=ws,
+            previous_task_status=previous_status,
         )
     await session.commit()
     await session.refresh(task)
@@ -196,7 +198,7 @@ async def cancel_task(
     task = await session.get(Task, task_id)
     if task is None:
         raise HTTPException(status_code=404, detail="Task not found")
-    task.status = TaskStatus.cancelled
+    transition_task_status(task, TaskStatus.cancelled)
     await session.commit()
     await session.refresh(task)
     return task

api/routers/workstreams.py

@@ -19,6 +19,7 @@ from api.schemas.workstream import (
     WorkstreamRead,
     WorkstreamUpdate,
 )
+from api.services.lifecycle import transition_workstream_status
 from api.workplan_status import (
     is_supported_workstream_status,
     normalize_workstream_status,
@@ -184,9 +185,13 @@ async def update_workstream(
     ws = await session.get(Workstream, workstream_id)
     if ws is None:
         raise HTTPException(status_code=404, detail="Workstream not found")
+    update_data = body.model_dump(exclude_unset=True)
+    status_update = update_data.pop("status", None)
     prev_status = ws.status
-    for field, value in body.model_dump(exclude_unset=True).items():
+    for field, value in update_data.items():
         setattr(ws, field, value)
+    if status_update is not None:
+        transition_workstream_status(ws, status_update)
     await session.commit()
     await session.refresh(ws)
 
@@ -216,7 +221,7 @@ async def archive_workstream(
     ws = await session.get(Workstream, workstream_id)
     if ws is None:
         raise HTTPException(status_code=404, detail="Workstream not found")
-    ws.status = "archived"
+    transition_workstream_status(ws, "archived")
     await session.commit()
     await session.refresh(ws)
     return ws

api/services/lifecycle.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import Any
 
 from api.workplan_status import normalize_workstream_status
@@ -11,6 +12,15 @@ TASK_ACTIVE_STATUSES = {"in_progress", "blocked"}
 PARENT_ACTIVATION_STATUSES = {"proposed", "ready", "backlog"}
 
 
+@dataclass(frozen=True)
+class LifecycleTransitionResult:
+    entity_type: str
+    previous_status: str
+    target_status: str
+    changed: bool
+    parent_activated: bool = False
+
+
 def status_value(status: Any) -> str:
     if hasattr(status, "value"):
         status = status.value
@@ -67,3 +77,49 @@ def activate_parent_for_task_start(
         return False
     parent_workstream.status = "active"
     return True
+
+
+def transition_workstream_status(
+    workstream: Any,
+    target_status: Any,
+) -> LifecycleTransitionResult:
+    """Apply a canonical workstream status transition."""
+    previous_status = normalize_workstream_status(getattr(workstream, "status", None))
+    normalised_target = normalize_workstream_status(target_status)
+    workstream.status = normalised_target
+    return LifecycleTransitionResult(
+        entity_type="workstream",
+        previous_status=previous_status,
+        target_status=normalised_target,
+        changed=previous_status != normalised_target,
+    )
+
+
+def transition_task_status(
+    task: Any,
+    target_status: Any,
+    *,
+    parent_workstream: Any = None,
+    previous_task_status: Any = None,
+    status_coercer: Any = None,
+) -> LifecycleTransitionResult:
+    """Apply a task status transition and activate the parent when work starts."""
+    previous_status = status_value(
+        getattr(task, "status", None)
+        if previous_task_status is None
+        else previous_task_status
+    )
+    normalised_target = status_value(target_status)
+    task.status = status_coercer(normalised_target) if status_coercer else target_status
+    parent_activated = activate_parent_for_task_start(
+        previous_task_status=previous_status,
+        new_task_status=normalised_target,
+        parent_workstream=parent_workstream,
+    )
+    return LifecycleTransitionResult(
+        entity_type="task",
+        previous_status=previous_status,
+        target_status=normalised_target,
+        changed=previous_status != normalised_target,
+        parent_activated=parent_activated,
+    )

tests/test_lifecycle.py

@@ -7,6 +7,8 @@ from api.services.lifecycle import (
     should_activate_parent_for_active_tasks,
     should_activate_parent_for_task_start,
     status_value,
+    transition_task_status,
+    transition_workstream_status,
 )
 
 
@@ -66,3 +68,34 @@ def test_status_value_unwraps_enum_like_values():
         value = "In_Progress"
 
     assert status_value(Status()) == "in_progress"
+
+
+def test_transition_workstream_status_normalizes_aliases():
+    class Workstream:
+        status = "todo"
+
+    ws = Workstream()
+    result = transition_workstream_status(ws, "done")
+
+    assert ws.status == "finished"
+    assert result.previous_status == "ready"
+    assert result.target_status == "finished"
+    assert result.changed is True
+
+
+def test_transition_task_status_activates_parent_once():
+    class Task:
+        status = "todo"
+
+    class Workstream:
+        status = "ready"
+
+    task = Task()
+    ws = Workstream()
+    result = transition_task_status(task, "in_progress", parent_workstream=ws)
+
+    assert task.status == "in_progress"
+    assert ws.status == "active"
+    assert result.parent_activated is True
+    assert result.previous_status == "todo"
+    assert result.target_status == "in_progress"

tests/test_routers_core.py

@@ -597,6 +597,58 @@ class TestReconciliationEndpoints:
         r = await client.get(f"/tasks/{task['id']}")
         assert r.json()["status"] == "in_progress"
 
+    async def test_apply_task_start_write_through_activates_parent_file_and_db(self, client, tmp_path):
+        await _create_domain(client)
+        repo_root = tmp_path / "repo"
+        workplans = repo_root / "workplans"
+        workplans.mkdir(parents=True)
+        repo = await _create_repo(client, local_path=repo_root)
+        topic = await _create_topic(client)
+        ws = await _create_workstream(client, topic["id"], repo_id=repo["id"], status="ready")
+        task = await _create_task(client, ws["id"])
+        wp = workplans / "STATE-WP-9999-demo.md"
+        wp.write_text(
+            "---\n"
+            "id: STATE-WP-9999\n"
+            "type: workplan\n"
+            "title: Demo\n"
+            "domain: custodian\n"
+            "repo: state-hub\n"
+            "status: ready\n"
+            f"state_hub_workstream_id: \"{ws['id']}\"\n"
+            "---\n\n"
+            "## Demo Task\n\n"
+            "```task\n"
+            "id: STATE-WP-9999-T01\n"
+            "status: todo\n"
+            "priority: high\n"
+            f"state_hub_task_id: \"{task['id']}\"\n"
+            "```\n",
+            encoding="utf-8",
+        )
+
+        r = await client.post("/reconciliation/state-change", json={
+            "target_type": "task",
+            "target_id": task["id"],
+            "target_status": "in_progress",
+            "actor": "dashboard",
+            "expected_current_status": "todo",
+            "apply": True,
+        })
+
+        assert r.status_code == 200, r.text
+        body = r.json()
+        assert body["write_through_result"] == "applied"
+        text = wp.read_text(encoding="utf-8")
+        assert "status: active" in text
+        assert "status: in_progress" in text
+
+        r = await client.get(f"/workstreams/{ws['id']}")
+        assert r.json()["status"] == "active"
+
+        r = await client.get(f"/tasks/{task['id']}")
+        assert r.json()["status"] == "in_progress"
+
     async def test_apply_task_confirmation_case_creates_reconciliation_message(self, client, tmp_path):
         await _create_domain(client)
         repo_root = tmp_path / "repo"

workplans/STATE-WP-0047-lifecycle-assertions-and-renormalization.md

@@ -4,7 +4,7 @@ type: workplan
 title: "Lifecycle Assertions and Renormalization"
 domain: custodian
 repo: state-hub
-status: active
+status: finished
 owner: codex
 topic_slug: custodian
 planning_priority: high
@@ -68,7 +68,7 @@ starts.
 
 ```task
 id: STATE-WP-0047-T02
-status: in_progress
+status: done
 priority: high
 state_hub_task_id: "56d9b6b9-fba1-4997-bdd5-875187cafa2d"
 ```
@@ -88,6 +88,13 @@ Progress 2026-05-23: consistency tooling now uses the shared lifecycle helper
 to detect and repair planning-state workplans with active tasks. Future UI
 actions still need to route through the shared transition layer.
 
+Result 2026-05-23: expanded the shared lifecycle helper into canonical
+workstream/task transition functions with result metadata. Direct task and
+workstream routes, flow advancement, consistency tooling, and UI-originated
+reconciliation now route status changes through the shared layer. Task-start
+write-through also patches parent workplan frontmatter to `active` when the
+shared helper activates the parent workstream.
+
 ## T03 - Auto-Advance Workstream On Task Start
 
 ```task