feat: add fast forge work-record reconciliation

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a053ff-1d6f-7fe2-ac1c-a6eb40a42a0c

api/main.py

@@ -26,6 +26,7 @@ from api.routers import legacy_meter
 from api.routers import review_contracts
 from api.routers import identifier_migrations
 from api.routers import repository_renames
+from api.routers import work_record_projection
 
 
 class ETagMiddleware(BaseHTTPMiddleware):
@@ -112,6 +113,7 @@ app.include_router(consistency_sweep.router)
 app.include_router(repos.router)
 app.include_router(repository_renames.router)
 app.include_router(repository_renames.operation_router)
+app.include_router(work_record_projection.router)
 app.include_router(topics.router)
 app.include_router(workstreams.router)
 app.include_router(workstreams.workplan_router)

api/routers/work_record_projection.py

@@ -0,0 +1,151 @@
+"""Primary-only fast path for one forge-derived repository projection."""
+
+from __future__ import annotations
+
+import asyncio
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.config import settings
+from api.database import get_session
+from api.models.task import Task
+from api.models.workplan import Workplan
+from api.models.workplan_dependency import WorkplanDependency
+from api.schemas.work_record_projection import (
+    RepositoryProjectionReconcile,
+    RepositoryProjectionSnapshot,
+)
+from api.services.forge_projection import (
+    ForgeDeriveError,
+    ForgeUnreadableError,
+    derive_from_forge,
+    reset_repository_projection,
+)
+from api.services.repository_aliases import resolve_repository_slug
+
+router = APIRouter(prefix="/repos", tags=["repository-work-record-projection"])
+
+
+@router.get(
+    "/{slug}/work-record-projection/snapshot",
+    response_model=RepositoryProjectionSnapshot,
+)
+async def repository_work_record_snapshot(
+    slug: str,
+    session: AsyncSession = Depends(get_session),
+) -> RepositoryProjectionSnapshot:
+    """Return all consistency-check inputs for one repository in one request."""
+    resolution = await resolve_repository_slug(session, slug)
+    workplans = list(
+        (
+            await session.execute(
+                select(Workplan)
+                .where(Workplan.repo_id == resolution.repo.id)
+                .order_by(Workplan.slug)
+            )
+        ).scalars()
+    )
+    workplan_ids = [workplan.id for workplan in workplans]
+    tasks: list[Task] = []
+    dependencies: list[WorkplanDependency] = []
+    if workplan_ids:
+        tasks = list(
+            (
+                await session.execute(
+                    select(Task)
+                    .where(Task.workplan_id.in_(workplan_ids))
+                    .order_by(Task.workplan_id, Task.id)
+                )
+            ).scalars()
+        )
+        dependencies = list(
+            (
+                await session.execute(
+                    select(WorkplanDependency)
+                    .where(WorkplanDependency.from_workplan_id.in_(workplan_ids))
+                    .order_by(
+                        WorkplanDependency.from_workplan_id, WorkplanDependency.id
+                    )
+                )
+            ).scalars()
+        )
+    return RepositoryProjectionSnapshot(
+        schema="state-hub.repository-projection-snapshot.v1",
+        repo_slug=resolution.canonical_slug,
+        repo_id=resolution.repo.id,
+        workplans=workplans,
+        tasks=tasks,
+        dependencies=dependencies,
+    )
+
+
+@router.post("/{slug}/work-record-projection/reconcile")
+async def reconcile_repository_work_records(
+    slug: str,
+    body: RepositoryProjectionReconcile,
+    session: AsyncSession = Depends(get_session),
+) -> dict:
+    """Derive an exact pushed commit centrally and apply it transactionally."""
+    if settings.state_hub_instance_role != "primary":
+        raise HTTPException(
+            status_code=409,
+            detail={
+                "message": "repository projection writes require the primary State Hub",
+                "instance_role": settings.state_hub_instance_role,
+                "instance_label": settings.state_hub_instance_label,
+            },
+        )
+
+    try:
+        derived = await asyncio.to_thread(derive_from_forge, slug)
+    except ForgeUnreadableError as exc:
+        raise HTTPException(
+            status_code=424,
+            detail={
+                "message": "repository is unreadable from the forge",
+                "detail": str(exc)[:300],
+            },
+        ) from exc
+    except ForgeDeriveError as exc:
+        raise HTTPException(
+            status_code=502,
+            detail={
+                "message": "forge projection derivation failed",
+                "detail": str(exc)[:300],
+            },
+        ) from exc
+
+    if derived.commit.lower() != body.expected_commit:
+        raise HTTPException(
+            status_code=409,
+            detail={
+                "message": "forge default branch is not at the expected commit",
+                "expected_commit": body.expected_commit,
+                "derived_commit": derived.commit,
+            },
+        )
+
+    outcome = await reset_repository_projection(
+        session,
+        slug,
+        acknowledge_retirements=body.acknowledge_retirements,
+        derived=derived,
+    )
+    if outcome.status in {"applied", "noop"} or outcome.released:
+        await session.commit()
+    else:
+        await session.rollback()
+
+    from api.routers.workstreams import _invalidate_workplan_index_cache
+
+    _invalidate_workplan_index_cache()
+    return {
+        "schema": "state-hub.repository-projection-reconcile.v1",
+        "instance_role": settings.state_hub_instance_role,
+        "instance_label": settings.state_hub_instance_label,
+        "expected_commit": body.expected_commit,
+        "derived_commit": derived.commit,
+        "outcome": outcome.to_dict(),
+    }

api/routers/workstreams.py

@@ -411,8 +411,11 @@ async def sync_workplan_bindings(
     """Upsert workstation workplan file bindings for remote API index fallback."""
     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()}
     for entry in body.bindings:
-        wp = await session.get(Workplan, entry.workplan_id)
+        wp = workplans.get(entry.workplan_id)
         if wp is None:
             continue
         wp.backing_filename = entry.filename
@@ -524,4 +527,4 @@ async def archive_workplan(
     workplan_id: uuid.UUID,
     session: AsyncSession = Depends(get_session),
 ) -> Workplan:
-    return await _archive_workplan(workplan_id=workplan_id, session=session)
+    return await _archive_workplan(workplan_id=workplan_id, session=session)

api/schemas/task.py

@@ -20,6 +20,7 @@ class TaskStatusMixin(BaseModel):
 
 class TaskCreate(TaskStatusMixin, WorkplanIdCreateMixin):
     id: uuid.UUID | None = None
+    record_id: str | None = None
     title: str
     description: str | None = None
     status: TaskStatus = TaskStatus.todo
@@ -100,6 +101,7 @@ class TaskStatusBulkSync(BaseModel):
 class TaskRead(TaskStatusMixin, WorkplanIdCompatMixin):
     model_config = ConfigDict(from_attributes=True)
     id: uuid.UUID
+    record_id: str | None = None
     title: str
     description: str | None = None
     status: TaskStatus

api/schemas/work_record_projection.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import re
+import uuid
+from typing import Literal
+
+from pydantic import BaseModel, Field, field_validator
+
+from api.schemas.task import TaskRead
+from api.schemas.workplan import WorkplanRead
+from api.schemas.workplan_dependency import WorkplanDependencyRead
+
+
+class RepositoryProjectionReconcile(BaseModel):
+    expected_commit: str
+    acknowledge_retirements: bool = False
+
+    @field_validator("expected_commit")
+    @classmethod
+    def validate_commit(cls, value: str) -> str:
+        value = value.strip().lower()
+        if not re.fullmatch(r"[0-9a-f]{40}", value):
+            raise ValueError("expected_commit must be a full 40-character Git SHA")
+        return value
+
+
+class RepositoryProjectionSnapshot(BaseModel):
+    """One bounded read of a repository's complete work-record projection."""
+
+    schema_version: Literal["state-hub.repository-projection-snapshot.v1"] = Field(
+        validation_alias="schema",
+        serialization_alias="schema",
+    )
+    repo_slug: str
+    repo_id: uuid.UUID
+    workplans: list[WorkplanRead]
+    tasks: list[TaskRead]
+    dependencies: list[WorkplanDependencyRead]

api/services/forge_projection.py

@@ -125,6 +125,10 @@ class DerivedTask:
     title: str | None
     status: str | None
     priority: str | None
+    description: str | None = None
+    needs_human: bool = False
+    intervention_note: str | None = None
+    blocking_reason: str | None = None
 
 
 @dataclass
@@ -135,6 +139,8 @@ class DerivedWorkplan:
     status: str | None
     relative_path: str
     archived: bool
+    owner: str | None = None
+    description: str | None = None
     tasks: list[DerivedTask] = field(default_factory=list)
 
 
@@ -180,6 +186,8 @@ class DerivedProjection:
                     "uuid": w.uuid,
                     "title": w.title,
                     "status": w.status,
+                    "owner": w.owner,
+                    "description": w.description,
                     "relative_path": w.relative_path,
                     "archived": w.archived,
                     "tasks": [
@@ -189,6 +197,10 @@ class DerivedProjection:
                             "title": t.title,
                             "status": t.status,
                             "priority": t.priority,
+                            "description": t.description,
+                            "needs_human": t.needs_human,
+                            "intervention_note": t.intervention_note,
+                            "blocking_reason": t.blocking_reason,
                         }
                         for t in w.tasks
                     ],
@@ -279,6 +291,11 @@ def _parse_tasks(body: str, workplan_id: str) -> list[DerivedTask]:
         if not title:
             prev = [t for pos, t in headings if pos < m.start()]
             title = prev[-1] if prev else None
+        following_headings = [pos for pos, _title in headings if pos > m.end()]
+        description_end = min(following_headings) if following_headings else len(body)
+        description = str(block.get("description") or "").strip()
+        if not description:
+            description = body[m.end() : description_end].strip()
         out.append(
             DerivedTask(
                 # A bare `T01` is not an identifier: it is unique only within
@@ -298,11 +315,36 @@ def _parse_tasks(body: str, workplan_id: str) -> list[DerivedTask]:
                 title=title,
                 status=(str(block["status"]).strip() if block.get("status") else None),
                 priority=(str(block["priority"]).strip() if block.get("priority") else None),
+                description=description or None,
+                needs_human=bool(block.get("needs_human", False)),
+                intervention_note=(
+                    str(block["intervention_note"]).strip()
+                    if block.get("intervention_note")
+                    else None
+                ),
+                blocking_reason=(
+                    str(block["blocking_reason"]).strip()
+                    if block.get("blocking_reason")
+                    else None
+                ),
             )
         )
     return out
 
 
+def _workplan_description(body: str) -> str | None:
+    """Return bounded prose under ``## Goal`` when one is present."""
+    match = re.search(r"^##\s+Goal\s*$", body, re.MULTILINE | re.IGNORECASE)
+    if match is None:
+        return None
+    remainder = body[match.end() :]
+    next_heading = re.search(r"^##\s+", remainder, re.MULTILINE)
+    if next_heading is not None:
+        remainder = remainder[: next_heading.start()]
+    value = remainder.strip()
+    return value[:4000] or None
+
+
 def derive_from_checkout(repo_root: Path, repo_slug: str, commit: str) -> DerivedProjection:
     """Derive a projection from an already-materialised checkout."""
     proj = DerivedProjection(repo_slug=repo_slug, commit=commit)
@@ -331,6 +373,12 @@ def derive_from_checkout(repo_root: Path, repo_slug: str, commit: str) -> Derive
                 status=(str(meta["status"]).strip() if meta.get("status") else None),
                 relative_path=str(path.relative_to(repo_root).as_posix()),
                 archived=path.parent.name == "archived",
+                owner=(str(meta["owner"]).strip() if meta.get("owner") else None),
+                description=(
+                    str(meta["description"]).strip()
+                    if meta.get("description")
+                    else _workplan_description(body)
+                ),
                 tasks=_parse_tasks(body, rid),
             )
         )
@@ -696,6 +744,10 @@ def _sync_existing_workplan_tasks(
                 "workplan_id": row.id,
                 "record_id": dt.record_id,
                 "title": (dt.title or dt.record_id),
+                "description": dt.description,
+                "needs_human": dt.needs_human,
+                "intervention_note": dt.intervention_note,
+                "blocking_reason": dt.blocking_reason,
             }
             st = _coerce_task_status(dt.status)
             if st is not None:
@@ -715,10 +767,30 @@ def _sync_existing_workplan_tasks(
         if dt.title and dt.title.strip() and ht.title != dt.title.strip():
             ht.title = dt.title.strip()
             changed = True
+        if getattr(ht, "description", None) != dt.description:
+            ht.description = dt.description
+            changed = True
         st = _coerce_task_status(dt.status)
         if st is not None and ht.status != st:
             ht.status = st
             changed = True
+        if getattr(ht, "needs_human", False) != dt.needs_human:
+            ht.needs_human = dt.needs_human
+            changed = True
+        if getattr(ht, "intervention_note", None) != dt.intervention_note:
+            ht.intervention_note = dt.intervention_note
+            changed = True
+        if getattr(ht, "blocking_reason", None) != dt.blocking_reason:
+            ht.blocking_reason = dt.blocking_reason
+            changed = True
+        if dt.priority:
+            try:
+                task_priority = TaskPriority(dt.priority.strip().lower())
+            except ValueError:
+                task_priority = None
+            if task_priority is not None and getattr(ht, "priority", None) != task_priority:
+                ht.priority = task_priority
+                changed = True
         if changed:
             outcome.updated_tasks.append(dt.record_id)
 
@@ -1054,7 +1126,9 @@ async def reset_repository_projection(
                 topic_id=repo.topic_id,
                 slug=w.record_id.lower(),
                 title=w.title or w.record_id,
+                description=w.description,
                 status=w.status or "proposed",
+                owner=w.owner,
                 backing_filename=w.relative_path.rsplit("/", 1)[-1],
                 backing_relative_path=w.relative_path,
                 backing_archived=w.archived,
@@ -1071,8 +1145,12 @@ async def reset_repository_projection(
                         workplan_id=row.id,
                         record_id=t.record_id,
                         title=t.title or t.record_id,
+                        description=t.description,
                         status=t.status or "todo",
                         priority=t.priority or "medium",
+                        needs_human=t.needs_human,
+                        intervention_note=t.intervention_note,
+                        blocking_reason=t.blocking_reason,
                     )
                 )
             outcome.created.append(w.record_id)
@@ -1090,6 +1168,12 @@ async def reset_repository_projection(
         if w.title and w.title.strip() and row.title != w.title.strip():
             row.title = w.title.strip()
             changed = True
+        if w.description and getattr(row, "description", None) != w.description:
+            row.description = w.description
+            changed = True
+        if getattr(row, "owner", None) != w.owner:
+            row.owner = w.owner
+            changed = True
         if w.status and row.status != w.status:
             row.status = w.status
             changed = True

api/services/write_idempotency.py

@@ -41,6 +41,12 @@ WRITE_ROUTE_RULES: tuple[WriteRouteRule, ...] = (
     WriteRouteRule("POST", r"/decisions", "append", "record decision"),
     WriteRouteRule("PATCH", r"/tasks/[^/]+", "replace", "update task"),
     WriteRouteRule("POST", r"/tasks/bulk-status-sync", "replace", "bulk task status sync"),
+    WriteRouteRule(
+        "POST",
+        r"/repos/[^/]+/work-record-projection/reconcile",
+        "replace",
+        "reconcile one forge-derived repository projection",
+    ),
     WriteRouteRule("PATCH", r"/decisions/[^/]+", "replace", "update decision"),
     WriteRouteRule("POST", r"/decisions/[^/]+/resolve", "replace", "resolve decision"),
     WriteRouteRule("PATCH", r"/workplans/[^/]+", "replace", "update workplan"),