Add lifecycle renormalization consistency repair

api/services/lifecycle.py

@@ -7,6 +7,7 @@ from api.workplan_status import normalize_workstream_status
 
 TASK_STARTED_STATUS = "in_progress"
 TASK_NOT_STARTED_STATUS = "todo"
+TASK_ACTIVE_STATUSES = {"in_progress", "blocked"}
 PARENT_ACTIVATION_STATUSES = {"proposed", "ready", "backlog"}
 
 
@@ -31,6 +32,24 @@ def should_activate_parent_for_task_start(
     )
 
 
+def has_active_task_status(task_statuses: list[Any] | tuple[Any, ...]) -> bool:
+    """Return whether any task status represents currently active work."""
+    return any(status_value(status) in TASK_ACTIVE_STATUSES for status in task_statuses)
+
+
+def should_activate_parent_for_active_tasks(
+    *,
+    parent_workstream_status: Any,
+    task_statuses: list[Any] | tuple[Any, ...],
+) -> bool:
+    """Return whether existing task state implies an active parent workstream."""
+    return (
+        normalize_workstream_status(parent_workstream_status)
+        in PARENT_ACTIVATION_STATUSES
+        and has_active_task_status(task_statuses)
+    )
+
+
 def activate_parent_for_task_start(
     *,
     previous_task_status: Any,

scripts/consistency_check.py

@@ -25,6 +25,7 @@ Checks:
   C-19  workstream-planning-drift WARN Yes  planning_priority/planning_order differs between file and DB
   C-20  workstream-dependency-missing WARN Yes Workplan dependency frontmatter missing from DB graph
   C-22  task-description-drift  WARN  Yes  Task description/content differs between file and DB
+  C-23  workstream-active-task-planning-status WARN Yes Workstream/workplan is planning while a task is in progress or blocked
 
 Usage:
     python scripts/consistency_check.py --repo SLUG [--fix] [--no-writeback] [--json] [--api-base URL]
@@ -65,6 +66,7 @@ from api.workplan_status import (  # noqa: E402
     normalize_workstream_status as _normalize_workstream_status,
     ready_review_status,
 )
+from api.services.lifecycle import should_activate_parent_for_active_tasks  # noqa: E402
 
 try:
     import yaml as _yaml
@@ -349,6 +351,34 @@ def _add_frontmatter_field(file_path: Path, key: str, value: str) -> None:
     file_path.write_text("\n".join(lines), encoding="utf-8")
 
 
+def _patch_frontmatter_field(file_path: Path, key: str, value: str) -> bool:
+    """Update or insert a scalar frontmatter field without rewriting the file."""
+    text = file_path.read_text(encoding="utf-8")
+    if not text.startswith("---"):
+        return False
+    lines = text.split("\n")
+    close_idx = None
+    for i, line in enumerate(lines[1:], 1):
+        if line.strip() == "---":
+            close_idx = i
+            break
+    if close_idx is None:
+        return False
+
+    new_line = f"{key}: {value}"
+    for i in range(1, close_idx):
+        if re.match(rf"^\s*{re.escape(key)}\s*:", lines[i]):
+            if lines[i] == new_line:
+                return False
+            lines[i] = new_line
+            file_path.write_text("\n".join(lines), encoding="utf-8")
+            return True
+
+    lines.insert(close_idx, new_line)
+    file_path.write_text("\n".join(lines), encoding="utf-8")
+    return True
+
+
 def _inject_task_id_into_block(
     file_path: Path, field_name: str, field_value: str, match_id: str
 ) -> bool:
@@ -711,11 +741,55 @@ def check_repo(api_base: str, repo_slug: str, repo_path_override: str | None = N
             )
             # Continue to check drift even with mismatched repo
 
+        tasks = get_tasks_from_workplan(meta, body)
+        db_tasks = _api_get(api_base, "/tasks", {"workstream_id": ws_id})
+        file_task_statuses = [
+            str(task.get("status", "")).strip()
+            for task in tasks
+            if not task.get("_parse_error")
+        ]
+        db_task_statuses = [
+            str(task.get("status", "")).strip()
+            for task in db_tasks
+            if isinstance(db_tasks, list)
+        ] if isinstance(db_tasks, list) else []
+        active_task_requires_activation = should_activate_parent_for_active_tasks(
+            parent_workstream_status=file_status,
+            task_statuses=[*file_task_statuses, *db_task_statuses],
+        )
+        if active_task_requires_activation:
+            report.add(
+                severity="WARN",
+                check_id="C-23",
+                message=(
+                    f"Lifecycle drift in '{ws.get('slug')}': workplan status "
+                    f"{file_status!r} has an active task — repair to 'active'"
+                ),
+                file_path=fname,
+                db_id=ws_id,
+                file_value=file_status,
+                db_value=ws.get("status", ""),
+                fixable=True,
+                _fix_context={
+                    "ws_id": ws_id,
+                    "wp_file": str(wp_file),
+                    "file_status": file_status,
+                    "db_status": ws.get("status", ""),
+                    "target_status": "active",
+                },
+            )
+
         # C-04: status drift — normalise file value before comparing so that
-        # legacy file/API aliases are not treated as drift.
+        # legacy file/API aliases are not treated as drift. Lifecycle repairs
+        # take precedence so a stale planning file cannot regress active work.
         db_status = ws.get("status", "")
         normalised_db_status = normalise_workstream_status(db_status)
-        if file_status and db_status and normalised_file_status != normalised_db_status:
+        if (
+            not active_task_requires_activation
+            and file_status
+            and db_status
+            and normalised_file_status != normalised_db_status
+        ):
             report.add(
                 severity="WARN", check_id="C-04",
                 message=(
@@ -806,8 +880,6 @@ def check_repo(api_base: str, repo_slug: str, repo_path_override: str | None = N
             )
 
         # C-10, C-11, C-12: task-level checks
-        tasks = get_tasks_from_workplan(meta, body)
-        db_tasks = _api_get(api_base, "/tasks", {"workstream_id": ws_id})
         db_task_by_id: dict[str, dict] = {}
         if isinstance(db_tasks, list):
             for t in db_tasks:
@@ -1209,7 +1281,11 @@ def _patch_task_status_in_file(
 
 
 def _git_commit_writeback(
-    repo_path: str, file_path: Path, changes: list[str]
+    repo_path: str,
+    file_path: Path,
+    changes: list[str],
+    *,
+    subject: str = "chore(consistency): sync task status from DB [auto]",
 ) -> bool:
     """Stage *file_path* and commit with a standard writeback message.
 
@@ -1219,7 +1295,7 @@ def _git_commit_writeback(
     from datetime import date as _date
     summary = "\n".join(f"  - {c}" for c in changes)
     msg = (
-        f"chore(consistency): sync task status from DB [auto]\n\n"
+        f"{subject}\n\n"
         f"Updated by fix-consistency on {_date.today().isoformat()}:\n"
         f"{summary}"
     )
@@ -1480,6 +1556,44 @@ def fix_repo(
                         f"{ctx['field']} → {ctx['value']!r}: {result['_error']}"
                     )
 
+            elif issue.check_id == "C-23":
+                ws_id = ctx["ws_id"]
+                target_status = ctx["target_status"]
+                result = _api_patch(api_base, f"/workstreams/{ws_id}", {"status": target_status})
+                if result is not None and "_error" not in result:
+                    report.fixes_applied.append(
+                        f"C-23 fixed: workstream {ws_id[:8]}… status → {target_status!r}"
+                    )
+                elif result is not None:
+                    report.fixes_applied.append(
+                        f"C-23 FAILED: workstream {ws_id[:8]}… "
+                        f"status → {target_status!r}: {result['_error']}"
+                    )
+
+                if no_writeback:
+                    report.fixes_applied.append(
+                        f"C-23 skipped file repair (--no-writeback): {Path(ctx['wp_file']).name}"
+                    )
+                else:
+                    wp_file = Path(ctx["wp_file"])
+                    old_status = ctx["file_status"]
+                    if _patch_frontmatter_field(wp_file, "status", target_status):
+                        committed = _git_commit_writeback(
+                            repo_path,
+                            wp_file,
+                            [f"workplan status: {old_status} → {target_status}"],
+                            subject="chore(consistency): renormalize lifecycle state [auto]",
+                        )
+                        suffix = " (committed)" if committed else " (file patched, commit failed)"
+                        report.fixes_applied.append(
+                            f"C-23 fixed: {wp_file.name} status "
+                            f"{old_status} → {target_status}{suffix}"
+                        )
+                    else:
+                        report.fixes_applied.append(
+                            f"C-23 SKIP: {wp_file.name} already has status {target_status!r}"
+                        )
+
             elif issue.check_id == "C-06":
                 wp_file = Path(ctx["wp_file"])
                 meta = ctx["meta"]

tests/test_consistency_check.py

@@ -28,10 +28,13 @@ from consistency_check import (
     _BACKGROUND_CHECKS,
     _detect_behind_remote,
     _git_pull,
+    _patch_frontmatter_field,
     _patch_task_status_in_file,
     _report_needs_action,
     archive_closed_workplans,
     canonical_workplan_filename,
+    check_repo,
+    fix_repo,
     get_tasks_from_workplan,
     iter_workplan_files,
     normalise_workstream_status,
@@ -726,6 +729,133 @@ class TestPatchTaskStatusInFile:
         assert result is False
 
 
+class TestPatchFrontmatterField:
+    def test_patches_existing_scalar_field(self, tmp_path):
+        f = tmp_path / "workplan.md"
+        f.write_text("---\nid: WP-001\nstatus: proposed\n---\nBody\n", encoding="utf-8")
+
+        assert _patch_frontmatter_field(f, "status", "active") is True
+
+        patched = f.read_text(encoding="utf-8")
+        assert "status: active" in patched
+        assert "id: WP-001" in patched
+
+    def test_is_idempotent_when_field_already_matches(self, tmp_path):
+        f = tmp_path / "workplan.md"
+        f.write_text("---\nid: WP-001\nstatus: active\n---\nBody\n", encoding="utf-8")
+
+        assert _patch_frontmatter_field(f, "status", "active") is False
+        assert f.read_text(encoding="utf-8").count("status: active") == 1
+
+
+class TestLifecycleRenormalization:
+    def _make_repo(self, tmp_path, status: str = "proposed") -> Path:
+        repo = tmp_path / "repo"
+        workplans = repo / "workplans"
+        workplans.mkdir(parents=True)
+        (workplans / "STATE-WP-0001-demo.md").write_text(
+            "---\n"
+            "id: STATE-WP-0001\n"
+            "type: workplan\n"
+            "title: Demo\n"
+            "domain: custodian\n"
+            "repo: state-hub\n"
+            f"status: {status}\n"
+            "owner: codex\n"
+            "state_hub_workstream_id: \"ws-1\"\n"
+            "---\n\n"
+            "## Implement Demo\n\n"
+            "```task\n"
+            "id: STATE-WP-0001-T01\n"
+            "status: in_progress\n"
+            "priority: high\n"
+            "state_hub_task_id: \"task-1\"\n"
+            "```\n",
+            encoding="utf-8",
+        )
+        return repo
+
+    def _api_get_for_repo(self, repo: Path):
+        ws = {
+            "id": "ws-1",
+            "repo_id": "repo-1",
+            "topic_id": "topic-1",
+            "slug": "state-wp-0001",
+            "title": "Demo",
+            "status": "proposed",
+            "planning_priority": None,
+            "planning_order": None,
+        }
+        task = {
+            "id": "task-1",
+            "title": "Implement Demo",
+            "status": "in_progress",
+            "description": None,
+        }
+
+        def fake_get(_api_base, path, params=None):
+            if path == "/repos/state-hub":
+                import socket
+
+                return {
+                    "id": "repo-1",
+                    "slug": "state-hub",
+                    "local_path": str(repo),
+                    "host_paths": {socket.gethostname(): str(repo)},
+                }
+            if path == "/workstreams/ws-1":
+                return ws
+            if path == "/tasks/task-1":
+                return task
+            if path == "/tasks" and params == {"workstream_id": "ws-1"}:
+                return [task]
+            if path == "/workstreams/ws-1/dependencies":
+                return []
+            if path == "/workstreams" and params == {"repo_id": "repo-1"}:
+                return [ws]
+            if path == "/workstreams" and params and params.get("topic_id") == "topic-1":
+                return []
+            return []
+
+        return fake_get
+
+    def test_active_task_in_planning_workplan_reports_c23_not_c04(self, tmp_path, monkeypatch):
+        repo = self._make_repo(tmp_path)
+        monkeypatch.setattr("consistency_check._api_get", self._api_get_for_repo(repo))
+
+        report = check_repo("http://unused", "state-hub")
+
+        check_ids = [issue.check_id for issue in report.issues]
+        assert "C-23" in check_ids
+        assert "C-04" not in check_ids
+        issue = next(issue for issue in report.issues if issue.check_id == "C-23")
+        assert issue.fixable is True
+        assert issue.file_value == "proposed"
+
+    def test_fix_repo_repairs_planning_workplan_with_active_task(self, tmp_path, monkeypatch):
+        repo = self._make_repo(tmp_path)
+        wp = repo / "workplans" / "STATE-WP-0001-demo.md"
+        patches = []
+
+        def fake_patch(_api_base, path, body):
+            patches.append((path, body))
+            return {"ok": True}
+
+        monkeypatch.setattr("consistency_check._api_get", self._api_get_for_repo(repo))
+        monkeypatch.setattr("consistency_check._api_patch", fake_patch)
+        monkeypatch.setattr("consistency_check._detect_behind_remote", lambda _repo_path: False)
+        monkeypatch.setattr("consistency_check._detect_ahead_of_remote", lambda _repo_path: 0)
+        monkeypatch.setattr("consistency_check._git_commit_writeback", lambda *args, **kwargs: True)
+        monkeypatch.setattr("consistency_check._write_custodian_brief", lambda *args, **kwargs: False)
+        monkeypatch.setattr("consistency_check._git_push", lambda _repo_path: (True, "pushed"))
+
+        report = fix_repo("http://unused", "state-hub")
+
+        assert ("/workstreams/ws-1", {"status": "active"}) in patches
+        assert "status: active" in wp.read_text(encoding="utf-8")
+        assert any("C-23 fixed" in fix for fix in report.fixes_applied)
+
+
 # ---------------------------------------------------------------------------
 # _git_pull (T02 remote fix helper)
 # ---------------------------------------------------------------------------

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

@@ -84,6 +84,10 @@ Progress 2026-05-23: added `api.services.lifecycle` with shared status
 normalization and parent-activation helpers. The task API now uses the helper;
 consistency tooling and future UI actions still need to adopt the shared layer.
 
+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.
+
 ## T03 - Auto-Advance Workstream On Task Start
 
 ```task
@@ -128,7 +132,7 @@ state entry assertions pass.
 
 ```task
 id: STATE-WP-0047-T05
-status: todo
+status: done
 priority: high
 state_hub_task_id: "611f0c22-34bc-494e-b520-068b4c3f0fec"
 ```
@@ -140,6 +144,12 @@ the `proposed workstream with in_progress task` case.
 Done when direct DB or file manipulation that breaks lifecycle invariants is
 caught by a repeatable repair path.
 
+Result 2026-05-23: added consistency rule C-23 for the
+`proposed`/`ready`/`backlog` workplan with an `in_progress` or `blocked` task
+case. The repair updates the DB workstream to `active`, patches workplan
+frontmatter to `status: active`, commits the writeback, and pushes through the
+existing consistency sync loop.
+
 ## T06 - Record Drift As Learning Input
 
 ```task
@@ -177,6 +187,10 @@ and consistency repair tests.
 Progress 2026-05-23: added engine and router coverage proving flow advancement
 honors current exit assertions before moving to the target workstation.
 
+Progress 2026-05-23: added consistency checker coverage for lifecycle
+renormalization detection and repair, including a guard that C-23 takes
+precedence over generic C-04 status drift.
+
 ## Acceptance Criteria
 
 - Starting task work deterministically activates the parent workstream.