fix(consistency): preserve immutable historical binding rows

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa

api/routers/workstreams.py

@@ -16,6 +16,8 @@ from api.database import get_session
 from api.events import EventEnvelope, publish_event
 from api.models.managed_repo import ManagedRepo
 from api.models.workplan import Workplan
+from api.models.task import Task
+from api.services.historical_projection import frozen_cluster_binding, validate_frozen_cluster_rows
 from api.schemas.workplan import (
     WorkplanBindingsSync,
     WorkplanCreate,
@@ -435,12 +437,42 @@ async def sync_workplan_bindings(
     session: AsyncSession = Depends(get_session),
 ) -> dict[str, int]:
     """Upsert workstation workplan file bindings for remote API index fallback."""
+    # Preflight the complete batch before assigning even an ordinary row.
+    try:
+        frozen = {entry.workplan_id: bound for entry in body.bindings
+                  if (bound := frozen_cluster_binding(entry.model_dump())) is not None}
+    except ValueError as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
+    requested_ids = {entry.workplan_id for entry in body.bindings}
+    rows = await session.execute(
+        select(Workplan).where(Workplan.id.in_(requested_ids))
+        .order_by(Workplan.id).with_for_update().execution_options(populate_existing=True)
+    )
+    workplans = {workplan.id: workplan for workplan in rows.scalars().all()}
+    if frozen:
+        repo_rows = await session.execute(
+            select(ManagedRepo).where(ManagedRepo.slug == "railiance-cluster")
+        )
+        repo = repo_rows.scalar_one_or_none()
+        task_rows = await session.execute(
+            select(Task).where(Task.workplan_id.in_(frozen))
+            .order_by(Task.id).with_for_update().execution_options(populate_existing=True)
+        )
+        tasks = task_rows.scalars().all()
+        try:
+            for workplan_id, bound in frozen.items():
+                validate_frozen_cluster_rows(
+                    bound, workplans.get(workplan_id), tasks, repo.id if repo else None
+                )
+        except ValueError as exc:
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
     synced_at = datetime.now(timezone.utc)
     updated = 0
-    requested_ids = {entry.workplan_id for entry in body.bindings}
-    rows = await session.execute(select(Workplan).where(Workplan.id.in_(requested_ids)))
-    workplans = {workplan.id: workplan for workplan in rows.scalars().all()}
+    preserved = 0
     for entry in body.bindings:
+        if entry.workplan_id in frozen:
+            preserved += 1
+            continue  # No setters, including backing_synced_at/updated_at.
         wp = workplans.get(entry.workplan_id)
         if wp is None:
             continue
@@ -449,9 +481,10 @@ async def sync_workplan_bindings(
         wp.backing_archived = entry.archived
         wp.backing_synced_at = synced_at
         updated += 1
-    await session.commit()
-    _invalidate_workplan_index_cache()
-    return {"updated": updated, "received": len(body.bindings)}
+    if updated:
+        await session.commit()
+        _invalidate_workplan_index_cache()
+    return {"updated": updated, "received": len(body.bindings), "preserved": preserved}
 
 
 @router.post("/", status_code=status.HTTP_410_GONE)

api/services/historical_projection.py

@@ -193,3 +193,46 @@ def bound_sources(repo_root: Path, repo_slug: str):
             raise ValueError("Historical source binding changed: " + entry["path"])
         result.append((entry, data.decode("utf-8")))
     return result
+
+
+# Only the three existing full-task Cluster bindings have immutable DB rows.
+# Other historical views retain their existing compatibility behavior.
+FROZEN_CLUSTER_BINDINGS = tuple(
+    entry for entry in BINDINGS
+    if entry["repo"] == "railiance-cluster" and "tasks" in entry
+)
+
+
+def frozen_cluster_binding(binding):
+    """Validate a claimed frozen binding; unrelated requests remain ordinary."""
+    matches = [entry for entry in FROZEN_CLUSTER_BINDINGS
+               if str(binding["workplan_id"]) == entry["uuid"]
+               or (binding["repo_slug"] == "railiance-cluster"
+                   and (binding["relative_path"] == entry["path"]
+                        or binding["filename"] == Path(entry["path"]).name))]
+    if not matches:
+        return None
+    if len(matches) != 1:
+        raise ValueError("frozen historical binding is ambiguous")
+    entry = matches[0]
+    expected = {
+        "workplan_id": entry["uuid"], "repo_slug": entry["repo"],
+        "filename": Path(entry["path"]).name, "relative_path": entry["path"],
+        "archived": True, "status": "finished",
+    }
+    if any((str(binding[key]) if key == "workplan_id" else binding[key]) != value
+           for key, value in expected.items()):
+        raise ValueError("frozen historical binding metadata changed")
+    return entry
+
+
+def validate_frozen_cluster_rows(entry, parent, tasks, repo_id):
+    """Receiving requires the exact existing terminal closure, never repair."""
+    if (parent is None or str(parent.id) != entry["uuid"]
+            or repo_id is None or parent.repo_id != repo_id
+            or parent.status not in {"done", "finished", "archived", "completed"}):
+        raise ValueError("frozen historical parent missing, foreign or nonterminal")
+    local = {str(task.id): task for task in tasks if task.workplan_id == parent.id}
+    if any(task["uuid"] not in local or local[task["uuid"]].status not in {"done", "cancel"}
+           for task in entry["tasks"]):
+        raise ValueError("frozen historical task missing, foreign or nonterminal")