Classify UI state reconciliation changes

2026-05-23 17:14:41 +02:00
parent 87d49a2573
commit 215a91e599
3 changed files with 294 additions and 2 deletions
--- a/api/services/reconciliation.py
+++ b/api/services/reconciliation.py
@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+from api.services.lifecycle import status_value
+from api.workplan_status import CLOSED_WORKSTREAM_STATUSES, normalize_workstream_status
+
+
+class ReconciliationClass(str, Enum):
+    WRITE_THROUGH = "write_through"
+    DEFERRED = "deferred"
+    HUMAN_CONFIRMATION = "human_confirmation"
+
+
+@dataclass(frozen=True)
+class StateChangeClassification:
+    reconciliation_class: ReconciliationClass
+    reason: str
+    follow_up: str
+
+
+WRITE_THROUGH_WORKSTREAM_STATUSES = {"proposed", "ready", "active", "backlog"}
+TASK_STATUSES = {"todo", "in_progress", "blocked", "done", "cancelled"}
+
+
+def classify_workstream_status_change(
+    *,
+    current_status: Any,
+    target_status: Any,
+    file_backed: bool,
+    archived_file: bool = False,
+    tasks_terminal: bool | None = None,
+) -> StateChangeClassification:
+    """Classify a UI-originated workstream status transition."""
+    current = normalize_workstream_status(current_status)
+    target = normalize_workstream_status(target_status)
+
+    if not file_backed:
+        return StateChangeClassification(
+            ReconciliationClass.DEFERRED,
+            "workstream is not backed by a local workplan file",
+            "create a reconciliation record for operator follow-up",
+        )
+    if archived_file:
+        return StateChangeClassification(
+            ReconciliationClass.HUMAN_CONFIRMATION,
+            "workplan file is archived",
+            "confirm whether to restore the file or leave the DB state unchanged",
+        )
+    if current == target:
+        return StateChangeClassification(
+            ReconciliationClass.WRITE_THROUGH,
+            "status is unchanged",
+            "no file update required",
+        )
+    if target in WRITE_THROUGH_WORKSTREAM_STATUSES and current not in CLOSED_WORKSTREAM_STATUSES:
+        return StateChangeClassification(
+            ReconciliationClass.WRITE_THROUGH,
+            "open lifecycle transition can be represented in workplan frontmatter",
+            "patch workplan status and sync the DB from file",
+        )
+    if target == "finished":
+        if tasks_terminal is True:
+            return StateChangeClassification(
+                ReconciliationClass.WRITE_THROUGH,
+                "all known tasks are terminal",
+                "patch workplan status to finished and sync the DB from file",
+            )
+        return StateChangeClassification(
+            ReconciliationClass.HUMAN_CONFIRMATION,
+            "finishing with non-terminal or unknown task state can hide open work",
+            "ask for confirmation or finish/cancel remaining tasks first",
+        )
+    if target == "archived":
+        return StateChangeClassification(
+            ReconciliationClass.HUMAN_CONFIRMATION,
+            "archiving may move files and hide active context",
+            "ask for confirmation and run the archive workflow",
+        )
+    if target == "blocked":
+        return StateChangeClassification(
+            ReconciliationClass.HUMAN_CONFIRMATION,
+            "blocked workstreams need an explicit blocker or dependency reason",
+            "capture the blocker before writing status",
+        )
+    return StateChangeClassification(
+        ReconciliationClass.DEFERRED,
+        f"unsupported workstream status transition {current!r} -> {target!r}",
+        "create a reconciliation record for operator review",
+    )
+
+
+def classify_task_status_change(
+    *,
+    current_status: Any,
+    target_status: Any,
+    file_backed: bool,
+    task_linked: bool,
+    archived_file: bool = False,
+    blocking_reason: str | None = None,
+) -> StateChangeClassification:
+    """Classify a UI-originated task status transition."""
+    current = status_value(current_status)
+    target = status_value(target_status)
+
+    if not file_backed:
+        return StateChangeClassification(
+            ReconciliationClass.DEFERRED,
+            "task belongs to a workstream without a local workplan file",
+            "create a reconciliation record for operator follow-up",
+        )
+    if archived_file:
+        return StateChangeClassification(
+            ReconciliationClass.HUMAN_CONFIRMATION,
+            "task belongs to an archived workplan file",
+            "confirm whether to restore or edit the archived file",
+        )
+    if not task_linked:
+        return StateChangeClassification(
+            ReconciliationClass.DEFERRED,
+            "task is not linked to a task block in the workplan file",
+            "create or repair the task block link before writing status",
+        )
+    if current == target:
+        return StateChangeClassification(
+            ReconciliationClass.WRITE_THROUGH,
+            "status is unchanged",
+            "no file update required",
+        )
+    if target == "blocked" and not (blocking_reason or "").strip():
+        return StateChangeClassification(
+            ReconciliationClass.HUMAN_CONFIRMATION,
+            "blocked tasks require a blocking reason",
+            "capture the blocker before writing status",
+        )
+    if target in TASK_STATUSES:
+        return StateChangeClassification(
+            ReconciliationClass.WRITE_THROUGH,
+            "task status can be represented in the workplan task block",
+            "patch task block status and sync the DB from file",
+        )
+    return StateChangeClassification(
+        ReconciliationClass.DEFERRED,
+        f"unsupported task status transition {current!r} -> {target!r}",
+        "create a reconciliation record for operator review",
+    )