Ship versioned Product Sessions

README.md

@@ -79,6 +79,10 @@ an instruction to an accepted result. It separates **work** from **acceptance**:
 - **The loop is an explicit state machine.** Phase, transition reason, and sequence are
   persisted rather than inferred from logs; recovery, the API, and the task UI therefore
   share the same semantics.
+- **Delivery can become the next loop.** Local-project tasks form a Product Session. A
+  verified v1 can accept a bug correction or changed product decision as an immutable
+  feedback delta, seed v2 from v1's exact verified patch, and retain both contracts,
+  Receipts, specifications, and change sets.
 
 If the evidence fails, the task does not become `completed` merely because the model
 called `finish`. Limit, stuck, cancelled, and error outcomes still receive an
@@ -213,6 +217,9 @@ and residual risks.
   bounded retries, dead-letter handling, and compare-and-update task claims.
 - Transactional local Git project edits through isolated clones and verified
   Apply/Discard/Undo change sets.
+- Product Sessions with content-addressed specifications, explicit implementation-fix
+  versus product-decision feedback, cumulative isolated revisions, latest-only Apply,
+  and a version navigator in the task UI.
 - File upload/download and `.xlsx`, `.docx`, `.csv`, image/vision workflows.
 - Signed capability-scoped skills, project/owner-scoped memory, and bounded
   Receipt-producing sub-agents.
@@ -266,7 +273,7 @@ make k8s-deployment-acceptance   # disposable production-mode cluster
 Current CI also audits locked Python/JavaScript dependencies, builds all runtime
 images, validates Compose/Kustomize boundaries, and packages/startup-smokes the
 desktop shell on macOS, Windows, and Ubuntu. Backend tests enforce a branch-aware
-70% coverage floor; the current full suite reports 72%.
+70% coverage floor; the current full suite reports 75%.
 
 ## Repository map
 

apps/api/alembic/versions/0012_product_sessions.py

@@ -0,0 +1,133 @@
+"""Add durable Product Sessions and immutable revision specifications.
+
+Revision ID: 0012_product_sessions
+Revises: 0011_loop_state_machine
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import uuid
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0012_product_sessions"
+down_revision: str | None = "0011_loop_state_machine"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _specification(goal: str, criteria: object) -> tuple[dict[str, object], str]:
+    rubric = criteria if isinstance(criteria, list) else []
+    specification: dict[str, object] = {
+        "schema": "loop.product-specification/v1",
+        "original_goal": goal.strip(),
+        "required_acceptance_criteria": rubric,
+        "feedback_history": [],
+        "previous_contract_hash": None,
+        "previous_receipt_hash": None,
+    }
+    canonical = json.dumps(
+        specification, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+    ).encode("utf-8")
+    return specification, hashlib.sha256(canonical).hexdigest()
+
+
+def upgrade() -> None:
+    op.create_table(
+        "product_sessions",
+        sa.Column("owner_id", sa.String(length=255), nullable=False),
+        sa.Column("project_id", sa.String(length=100), nullable=False),
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_product_sessions")),
+    )
+    op.create_index(op.f("ix_product_sessions_owner_id"), "product_sessions", ["owner_id"])
+    op.create_index(op.f("ix_product_sessions_project_id"), "product_sessions", ["project_id"])
+    op.add_column("tasks", sa.Column("product_session_id", sa.Uuid(), nullable=True))
+    op.add_column("tasks", sa.Column("product_revision", sa.Integer(), nullable=True))
+    op.add_column("tasks", sa.Column("previous_revision_id", sa.Uuid(), nullable=True))
+    op.add_column("tasks", sa.Column("superseded_by_id", sa.Uuid(), nullable=True))
+    op.add_column("tasks", sa.Column("feedback_kind", sa.String(length=30), nullable=True))
+    op.add_column("tasks", sa.Column("feedback_delta", sa.Text(), nullable=True))
+    op.add_column("tasks", sa.Column("product_specification", sa.JSON(), nullable=True))
+    op.add_column("tasks", sa.Column("specification_hash", sa.String(length=64), nullable=True))
+    op.create_index(op.f("ix_tasks_product_session_id"), "tasks", ["product_session_id"])
+    op.create_index(
+        "uq_tasks_product_session_id_product_revision",
+        "tasks",
+        ["product_session_id", "product_revision"],
+        unique=True,
+    )
+
+    connection = op.get_bind()
+    product_sessions = sa.table(
+        "product_sessions",
+        sa.column("id", sa.Uuid()),
+        sa.column("owner_id", sa.String()),
+        sa.column("project_id", sa.String()),
+    )
+    tasks = sa.table(
+        "tasks",
+        sa.column("id", sa.Uuid()),
+        sa.column("product_session_id", sa.Uuid()),
+        sa.column("product_revision", sa.Integer()),
+        sa.column("product_specification", sa.JSON()),
+        sa.column("specification_hash", sa.String()),
+    )
+    rows = connection.execute(
+        sa.text(
+            "SELECT id, owner_id, project_id, goal, rubric FROM tasks "
+            "WHERE project_source_path IS NOT NULL AND parent_id IS NULL"
+        )
+    ).mappings()
+    for row in rows:
+        session_id = uuid.uuid4()
+        raw_rubric = row["rubric"]
+        if isinstance(raw_rubric, str):
+            try:
+                raw_rubric = json.loads(raw_rubric)
+            except ValueError:
+                raw_rubric = []
+        specification, digest = _specification(str(row["goal"]), raw_rubric)
+        connection.execute(
+            product_sessions.insert().values(
+                id=session_id,
+                owner_id=row["owner_id"],
+                project_id=row["project_id"],
+            )
+        )
+        connection.execute(
+            tasks.update()
+            .where(tasks.c.id == uuid.UUID(str(row["id"])))
+            .values(
+                product_session_id=session_id,
+                product_revision=1,
+                product_specification=specification,
+                specification_hash=digest,
+            )
+        )
+
+
+def downgrade() -> None:
+    op.drop_index("uq_tasks_product_session_id_product_revision", table_name="tasks")
+    op.drop_index(op.f("ix_tasks_product_session_id"), table_name="tasks")
+    op.drop_column("tasks", "specification_hash")
+    op.drop_column("tasks", "product_specification")
+    op.drop_column("tasks", "feedback_delta")
+    op.drop_column("tasks", "feedback_kind")
+    op.drop_column("tasks", "superseded_by_id")
+    op.drop_column("tasks", "previous_revision_id")
+    op.drop_column("tasks", "product_revision")
+    op.drop_column("tasks", "product_session_id")
+    op.drop_index(op.f("ix_product_sessions_project_id"), table_name="product_sessions")
+    op.drop_index(op.f("ix_product_sessions_owner_id"), table_name="product_sessions")
+    op.drop_table("product_sessions")

apps/api/app/api/v1/routes/tasks.py

@@ -36,6 +36,7 @@
 from app.schemas.task import (
     ChangeSetRead,
     LimitDefaults,
+    ProductRevisionCreate,
     RespondIn,
     TaskCreate,
     TaskRead,
@@ -144,6 +145,34 @@ async def retry_task(
     return TaskRead.from_model(task)
 
 
+@router.get(
+    "/{task_id}/revisions",
+    response_model=list[TaskRead],
+    summary="List every immutable delivery in a Product Session",
+)
+async def product_revisions(task_id: uuid.UUID, service: TaskServiceDep) -> list[TaskRead]:
+    return [TaskRead.from_model(task) for task in await service.list_revisions(task_id)]
+
+
+@router.post(
+    "/{task_id}/revisions",
+    response_model=TaskRead,
+    status_code=status.HTTP_201_CREATED,
+    summary="Turn verified-delivery feedback into the next Product Session revision",
+    dependencies=[rate_limit(limit=10, window_seconds=60)],
+)
+async def create_product_revision(
+    task_id: uuid.UUID,
+    payload: ProductRevisionCreate,
+    service: TaskServiceDep,
+    background: BackgroundTasks,
+) -> TaskRead:
+    task = await service.create_revision(task_id, payload)
+    if payload.autostart:
+        background.add_task(trigger_task, task.id)
+    return TaskRead.from_model(task)
+
+
 @router.get("/{task_id}", response_model=TaskRead, summary="Get a task")
 async def get_task(task_id: uuid.UUID, service: TaskServiceDep) -> TaskRead:
     task = await service.get(task_id)

apps/api/app/db/models/__init__.py

@@ -1,7 +1,8 @@
 """Import all models here so Alembic's autogenerate sees them via ``Base.metadata``."""
 
+from app.db.models.product_session import ProductSessionModel
 from app.db.models.step import StepModel
 from app.db.models.task import TaskModel
 from app.db.models.trigger import TriggerModel
 
-__all__ = ["StepModel", "TaskModel", "TriggerModel"]
+__all__ = ["ProductSessionModel", "StepModel", "TaskModel", "TriggerModel"]

apps/api/app/db/models/product_session.py

@@ -0,0 +1,15 @@
+"""A durable product-delivery lineage across successive verified tasks."""
+
+from __future__ import annotations
+
+from sqlalchemy import String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base, TimestampMixin, UUIDPrimaryKeyMixin
+
+
+class ProductSessionModel(UUIDPrimaryKeyMixin, TimestampMixin, Base):
+    __tablename__ = "product_sessions"
+
+    owner_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
+    project_id: Mapped[str] = mapped_column(String(100), nullable=False, index=True)

apps/api/app/db/models/task.py

@@ -10,7 +10,7 @@
 import uuid
 from typing import Any
 
-from sqlalchemy import JSON, Boolean, Integer, String, Text, UniqueConstraint, Uuid
+from sqlalchemy import JSON, Boolean, Index, Integer, String, Text, UniqueConstraint, Uuid
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base, TimestampMixin, UUIDPrimaryKeyMixin
@@ -20,7 +20,15 @@
 
 class TaskModel(UUIDPrimaryKeyMixin, TimestampMixin, Base):
     __tablename__ = "tasks"
-    __table_args__ = (UniqueConstraint("owner_id", "idempotency_key"),)
+    __table_args__ = (
+        UniqueConstraint("owner_id", "idempotency_key"),
+        Index(
+            "uq_tasks_product_session_id_product_revision",
+            "product_session_id",
+            "product_revision",
+            unique=True,
+        ),
+    )
 
     goal: Mapped[str] = mapped_column(Text, nullable=False)
     owner_id: Mapped[str] = mapped_column(String(255), nullable=False, default="local", index=True)
@@ -73,6 +81,16 @@ class TaskModel(UUIDPrimaryKeyMixin, TimestampMixin, Base):
     # Sub-agent delegation: a spawned task points at its parent and tracks depth.
     parent_id: Mapped[uuid.UUID | None] = mapped_column(Uuid(), nullable=True, index=True)
     depth: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    # Product revision lineage is separate from parent_id, which exclusively
+    # represents the sub-agent execution tree.
+    product_session_id: Mapped[uuid.UUID | None] = mapped_column(Uuid(), nullable=True, index=True)
+    product_revision: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    previous_revision_id: Mapped[uuid.UUID | None] = mapped_column(Uuid(), nullable=True)
+    superseded_by_id: Mapped[uuid.UUID | None] = mapped_column(Uuid(), nullable=True)
+    feedback_kind: Mapped[str | None] = mapped_column(String(30), nullable=True)
+    feedback_delta: Mapped[str | None] = mapped_column(Text, nullable=True)
+    product_specification: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
+    specification_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
     # Chat origin: the channel/chat this task came from, so replies route back.
     chat_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     # The action awaiting approval while paused: {"tool": ..., "args": {...}}.

apps/api/app/repositories/product_session.py

@@ -0,0 +1,26 @@
+"""Data access for durable product-delivery lineages."""
+
+from __future__ import annotations
+
+import uuid
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models.product_session import ProductSessionModel
+from app.repositories.base import BaseRepository
+
+
+class ProductSessionRepository(BaseRepository[ProductSessionModel]):
+    model = ProductSessionModel
+
+    def __init__(self, session: AsyncSession) -> None:
+        super().__init__(session)
+
+    async def get_for_update(self, session_id: uuid.UUID) -> ProductSessionModel | None:
+        result: ProductSessionModel | None = await self.session.scalar(
+            select(ProductSessionModel)
+            .where(ProductSessionModel.id == session_id)
+            .with_for_update()
+        )
+        return result

apps/api/app/repositories/task.py

@@ -4,7 +4,7 @@
 
 import uuid
 
-from sqlalchemy import case, func, select, update
+from sqlalchemy import case, func, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.sql.elements import ColumnElement
 
@@ -24,7 +24,10 @@ async def list_roots(
     ) -> list[TaskModel]:
         """Top-level tasks only — spawned sub-agents (parent_id set) are excluded
         so they don't pollute the task list; they show under their parent."""
-        conditions: list[ColumnElement[bool]] = [TaskModel.parent_id.is_(None)]
+        conditions: list[ColumnElement[bool]] = [
+            TaskModel.parent_id.is_(None),
+            or_(TaskModel.product_session_id.is_(None), TaskModel.superseded_by_id.is_(None)),
+        ]
         if owner_id is not None:
             conditions.append(TaskModel.owner_id == owner_id)
         stmt = (
@@ -78,7 +81,10 @@ async def claim_pending(self, task_id: uuid.UUID) -> TaskModel | None:
         return task
 
     async def count_roots(self, *, owner_id: str | None = None) -> int:
-        conditions: list[ColumnElement[bool]] = [TaskModel.parent_id.is_(None)]
+        conditions: list[ColumnElement[bool]] = [
+            TaskModel.parent_id.is_(None),
+            or_(TaskModel.product_session_id.is_(None), TaskModel.superseded_by_id.is_(None)),
+        ]
         if owner_id is not None:
             conditions.append(TaskModel.owner_id == owner_id)
         stmt = select(func.count()).select_from(TaskModel).where(*conditions)
@@ -106,6 +112,14 @@ async def list_children(self, parent_id: uuid.UUID) -> list[TaskModel]:
         )
         return list((await self.session.scalars(stmt)).all())
 
+    async def list_revisions(self, session_id: uuid.UUID) -> list[TaskModel]:
+        stmt = (
+            select(TaskModel)
+            .where(TaskModel.product_session_id == session_id)
+            .order_by(TaskModel.product_revision.asc())
+        )
+        return list((await self.session.scalars(stmt)).all())
+
     async def list_descendants(self, parent_id: uuid.UUID) -> list[TaskModel]:
         descendants: list[TaskModel] = []
         frontier = [parent_id]