Add reconciliation state-change API contract

api/main.py

@@ -16,6 +16,7 @@ from api.routers import token_events
 from api.routers import interface_changes
 from api.routers import flows
 from api.routers import recently_on_scope
+from api.routers import reconciliation
 
 
 class ETagMiddleware(BaseHTTPMiddleware):
@@ -100,6 +101,7 @@ app.include_router(tpsc.router)
 app.include_router(token_events.router)
 app.include_router(interface_changes.router)
 app.include_router(flows.router)
+app.include_router(reconciliation.router)
 app.include_router(state.router)
 app.include_router(policy.router)
 

api/routers/reconciliation.py

@@ -0,0 +1,99 @@
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.database import get_session
+from api.models.task import Task
+from api.models.workstream import Workstream
+from api.schemas.reconciliation import StateChangeRequest, StateChangeResponse
+from api.services.lifecycle import status_value
+from api.services.reconciliation import (
+    classify_task_status_change,
+    classify_workstream_status_change,
+)
+from api.workplan_status import normalize_workstream_status
+
+router = APIRouter(prefix="/reconciliation", tags=["reconciliation"])
+
+
+def _bool_or_default(value: bool | None, default: bool) -> bool:
+    return default if value is None else value
+
+
+async def _workstream_tasks_terminal(session: AsyncSession, workstream_id: uuid.UUID) -> bool:
+    result = await session.execute(select(Task.status).where(Task.workstream_id == workstream_id))
+    statuses = [status_value(row[0]) for row in result.all()]
+    return bool(statuses) and all(status in {"done", "cancelled"} for status in statuses)
+
+
+@router.post("/state-change", response_model=StateChangeResponse)
+async def classify_state_change(
+    body: StateChangeRequest,
+    session: AsyncSession = Depends(get_session),
+) -> StateChangeResponse:
+    if body.target_type == "workstream":
+        ws = await session.get(Workstream, body.target_id)
+        if ws is None:
+            raise HTTPException(status_code=404, detail="Workstream not found")
+
+        file_backed = _bool_or_default(body.file_backed, ws.repo_id is not None)
+        archived_file = _bool_or_default(body.archived_file, False)
+        tasks_terminal = (
+            body.tasks_terminal
+            if body.tasks_terminal is not None
+            else await _workstream_tasks_terminal(session, ws.id)
+        )
+        classification = classify_workstream_status_change(
+            current_status=ws.status,
+            target_status=body.target_status,
+            file_backed=file_backed,
+            archived_file=archived_file,
+            tasks_terminal=tasks_terminal,
+        )
+        return StateChangeResponse(
+            target_type=body.target_type,
+            target_id=body.target_id,
+            actor=body.actor,
+            intent=body.intent,
+            current_status=normalize_workstream_status(ws.status),
+            target_status=normalize_workstream_status(body.target_status),
+            file_backed=file_backed,
+            archived_file=archived_file,
+            tasks_terminal=tasks_terminal,
+            reconciliation_class=classification.reconciliation_class,
+            reason=classification.reason,
+            follow_up=classification.follow_up,
+        )
+
+    task = await session.get(Task, body.target_id)
+    if task is None:
+        raise HTTPException(status_code=404, detail="Task not found")
+
+    ws = await session.get(Workstream, task.workstream_id)
+    file_backed = _bool_or_default(body.file_backed, bool(ws and ws.repo_id))
+    archived_file = _bool_or_default(body.archived_file, False)
+    task_linked = _bool_or_default(body.task_linked, True)
+    classification = classify_task_status_change(
+        current_status=task.status,
+        target_status=body.target_status,
+        file_backed=file_backed,
+        archived_file=archived_file,
+        task_linked=task_linked,
+        blocking_reason=body.blocking_reason,
+    )
+    return StateChangeResponse(
+        target_type=body.target_type,
+        target_id=body.target_id,
+        actor=body.actor,
+        intent=body.intent,
+        current_status=status_value(task.status),
+        target_status=status_value(body.target_status),
+        file_backed=file_backed,
+        archived_file=archived_file,
+        task_linked=task_linked,
+        reconciliation_class=classification.reconciliation_class,
+        reason=classification.reason,
+        follow_up=classification.follow_up,
+    )

api/schemas/reconciliation.py

@@ -0,0 +1,39 @@
+import uuid
+from typing import Literal
+
+from pydantic import BaseModel
+
+from api.services.reconciliation import ReconciliationClass
+
+
+TargetType = Literal["workstream", "task"]
+
+
+class StateChangeRequest(BaseModel):
+    target_type: TargetType
+    target_id: uuid.UUID
+    target_status: str
+    actor: str = "dashboard"
+    intent: str | None = None
+    file_backed: bool | None = None
+    archived_file: bool | None = None
+    task_linked: bool | None = None
+    tasks_terminal: bool | None = None
+    blocking_reason: str | None = None
+
+
+class StateChangeResponse(BaseModel):
+    target_type: TargetType
+    target_id: uuid.UUID
+    actor: str
+    intent: str | None = None
+    current_status: str
+    target_status: str
+    file_backed: bool
+    archived_file: bool
+    task_linked: bool | None = None
+    tasks_terminal: bool | None = None
+    reconciliation_class: ReconciliationClass
+    reason: str
+    follow_up: str
+    write_through_result: Literal["not_attempted"] = "not_attempted"

tests/test_routers_core.py

@@ -392,3 +392,75 @@ class TestFlowEndpoints:
 
         r = await client.get(f"/workstreams/{ws['id']}")
         assert r.json()["status"] == "active"
+
+
+class TestReconciliationEndpoints:
+    async def test_classify_workstream_open_transition_write_through(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        ws = await _create_workstream(client, topic["id"])
+
+        r = await client.post("/reconciliation/state-change", json={
+            "target_type": "workstream",
+            "target_id": ws["id"],
+            "target_status": "backlog",
+            "actor": "dashboard",
+            "intent": "push back to backlog",
+            "file_backed": True,
+        })
+
+        assert r.status_code == 200
+        body = r.json()
+        assert body["current_status"] == "active"
+        assert body["target_status"] == "backlog"
+        assert body["reconciliation_class"] == "write_through"
+        assert body["write_through_result"] == "not_attempted"
+        assert body["intent"] == "push back to backlog"
+
+    async def test_classify_workstream_finish_with_open_task_needs_confirmation(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        ws = await _create_workstream(client, topic["id"])
+        await _create_task(client, ws["id"])
+
+        r = await client.post("/reconciliation/state-change", json={
+            "target_type": "workstream",
+            "target_id": ws["id"],
+            "target_status": "finished",
+            "file_backed": True,
+        })
+
+        assert r.status_code == 200
+        body = r.json()
+        assert body["tasks_terminal"] is False
+        assert body["reconciliation_class"] == "human_confirmation"
+        assert "open work" in body["reason"]
+
+    async def test_classify_task_blocked_without_reason_needs_confirmation(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"])
+
+        r = await client.post("/reconciliation/state-change", json={
+            "target_type": "task",
+            "target_id": task["id"],
+            "target_status": "blocked",
+            "file_backed": True,
+            "task_linked": True,
+        })
+
+        assert r.status_code == 200
+        body = r.json()
+        assert body["current_status"] == "todo"
+        assert body["reconciliation_class"] == "human_confirmation"
+        assert "blocking reason" in body["reason"]
+
+    async def test_classify_unknown_workstream_returns_404(self, client):
+        r = await client.post("/reconciliation/state-change", json={
+            "target_type": "workstream",
+            "target_id": "00000000-0000-0000-0000-000000000001",
+            "target_status": "active",
+        })
+
+        assert r.status_code == 404

workplans/STATE-WP-0048-ui-state-change-reconciliation.md

@@ -72,7 +72,7 @@ pin the write-through, deferred, and human-confirmation decisions.
 
 ```task
 id: STATE-WP-0048-T02
-status: todo
+status: in_progress
 priority: high
 state_hub_task_id: "50c20ddf-f039-418b-a763-7a8f581be5b0"
 ```
@@ -84,6 +84,12 @@ and follow-up action.
 Done when dashboard actions no longer perform ambiguous direct status patches
 without reconciliation metadata.
 
+Progress 2026-05-23: added a classify-only
+`POST /reconciliation/state-change` API contract. It returns actor, intent,
+current/target status, file-backed flags, reconciliation class, reason,
+follow-up action, and `write_through_result: not_attempted`. Dashboard wiring
+and write-through execution remain for the next slice.
+
 ## T03 - Implement File Write-Through For Safe Changes
 
 ```task