migrations/versions/0006_album_proposals.py

@@ -0,0 +1,59 @@
+"""Versioned album naming proposals (US03-03).
+
+Revision ID: 0006_album_proposals
+Revises: 0005_safety_analysis
+Create Date: 2026-08-15
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0006_album_proposals"
+down_revision = "0005_safety_analysis"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "album_proposals",
+        sa.Column("id", sa.String(), primary_key=True),
+        # The album label (US03-01 identity) this proposal names. One current
+        # proposal per album; superseded ones keep their history in the events.
+        sa.Column("album", sa.String(), nullable=False),
+        sa.Column("proposed_name", sa.String(), nullable=True),
+        sa.Column("final_name", sa.String(), nullable=True),
+        sa.Column("rationale", sa.String(), nullable=True),
+        sa.Column("confidence", sa.Float(), nullable=True),
+        # pending | proposed | edited | approved | error
+        sa.Column("status", sa.String(), nullable=False, server_default="pending"),
+        sa.Column("error_code", sa.String(), nullable=True),
+        sa.Column("error_message", sa.String(), nullable=True),
+        # The US03-01 evidence content hash the proposal was generated from; a
+        # different current value means the proposal is stale.
+        sa.Column("evidence_version", sa.String(), nullable=True),
+        sa.Column("model", sa.String(), nullable=True),
+        sa.Column("prompt_version", sa.String(), nullable=True),
+        sa.Column("raw_response", sa.String(), nullable=True),
+        # Optimistic concurrency: an edit/approval must present this version.
+        sa.Column("version", sa.Integer(), nullable=False, server_default="1"),
+        sa.Column("approved_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+        ),
+    )
+    op.create_index("ix_album_proposals_album", "album_proposals", ["album"], unique=True)
+    op.create_index("ix_album_proposals_status", "album_proposals", ["status"])
+
+
+def downgrade() -> None:
+    op.drop_table("album_proposals")

photo_pipeline/api/app.py

@@ -15,6 +15,7 @@ from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles
 
 from photo_pipeline.api.routes import (
+    albums,
     analysis,
     duplicates,
     health,
@@ -63,6 +64,7 @@ def create_app(config: Config | None = None) -> FastAPI:
     app.include_router(safety.router, prefix="/api/v1")
     app.include_router(analysis.router, prefix="/api/v1")
     app.include_router(library.router, prefix="/api/v1")
+    app.include_router(albums.router, prefix="/api/v1")
     # Static single-page app (hash-routed). Mounted last so /api/v1 wins.
     if FRONTEND_DIR.is_dir():
         app.mount("/app", StaticFiles(directory=FRONTEND_DIR, html=True), name="app")

photo_pipeline/api/routes/albums.py

@@ -0,0 +1,84 @@
+"""Album evidence and naming-proposal API (US03-01/US03-03).
+
+Proposals are read, edited, and approved by album label. Edits and approvals carry an
+``expected_version``; a stale version returns 409 without mutating anything. Approval
+records intent only — no rename happens in this phase.
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Query, Request
+from fastapi.responses import JSONResponse
+
+from photo_pipeline.schemas import (
+    ApproveProposalRequest,
+    EditProposalRequest,
+    GenerateProposalsRequest,
+)
+from photo_pipeline.services.albums import AlbumService
+from photo_pipeline.services.proposals import ConflictError, ProposalError, ProposalService
+
+router = APIRouter(tags=["albums"])
+
+
+def _roots(request: Request) -> tuple:
+    return tuple(request.app.state.config.library_roots)
+
+
+def _albums(request: Request) -> AlbumService:
+    return AlbumService(request.app.state.session_factory, library_roots=_roots(request))
+
+
+def _proposals(request: Request) -> ProposalService:
+    return ProposalService(request.app.state.session_factory, library_roots=_roots(request))
+
+
+def _error(status: int, code: str, message: str) -> JSONResponse:
+    return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
+
+
+@router.get("/albums/evidence")
+def evidence(
+    request: Request,
+    offset: int = Query(0, ge=0),
+    limit: int = Query(50, ge=1, le=500),
+) -> dict:
+    return _albums(request).aggregate_evidence(offset=offset, limit=limit)
+
+
+@router.post("/albums/proposals")
+def generate(body: GenerateProposalsRequest, request: Request) -> dict:
+    return _proposals(request).generate(body.albums)
+
+
+@router.get("/albums/proposals")
+def list_proposals(request: Request, status: str | None = None) -> dict:
+    return _proposals(request).list(status=status)
+
+
+@router.get("/albums/proposals/{album:path}")
+def get_proposal(album: str, request: Request):
+    data = _proposals(request).get(album)
+    if data is None:
+        return _error(404, "not_found", f"no proposal for album {album!r}")
+    return data
+
+
+@router.post("/albums/proposals/{album:path}/edit")
+def edit_proposal(album: str, body: EditProposalRequest, request: Request):
+    try:
+        return _proposals(request).edit(album, body.name, expected_version=body.expected_version)
+    except ConflictError as error:
+        return _error(409, "version_conflict", str(error))
+    except ProposalError as error:
+        return _error(422, "invalid_proposal", str(error))
+
+
+@router.post("/albums/proposals/{album:path}/approve")
+def approve_proposal(album: str, body: ApproveProposalRequest, request: Request):
+    try:
+        return _proposals(request).approve(album, expected_version=body.expected_version)
+    except ConflictError as error:
+        return _error(409, "version_conflict", str(error))
+    except ProposalError as error:
+        return _error(422, "invalid_proposal", str(error))

photo_pipeline/models/__init__.py

@@ -4,6 +4,7 @@ Importing this package registers every model on ``Base.metadata``, which the
 Alembic environment relies on.
 """
 
+from photo_pipeline.models.albums import AlbumProposal
 from photo_pipeline.models.assets import Asset, AssetPath
 from photo_pipeline.models.duplicates import (
     DuplicateCluster,
@@ -15,6 +16,7 @@ from photo_pipeline.models.thumbnails import Thumbnail
 from photo_pipeline.models.workflow import AnalysisResult, SafetyReview
 
 __all__ = [
+    "AlbumProposal",
     "Asset",
     "AssetPath",
     "DuplicateCluster",

photo_pipeline/models/albums.py

@@ -0,0 +1,49 @@
+"""Album naming proposal persistence (US03-03).
+
+One current proposal row per album label (the US03-01 identity). The row records
+what was proposed, from which evidence version and model/prompt version, the user's
+edit, the approval state, and an optimistic ``version`` that every edit/approval
+must present so stale browser state cannot overwrite a newer decision.
+
+Approval records intent only — no rename happens here (that is Phase D).
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, Float, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from photo_pipeline.db import Base
+
+
+class AlbumProposal(Base):
+    __tablename__ = "album_proposals"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    album: Mapped[str] = mapped_column(String, nullable=False, unique=True, index=True)
+
+    proposed_name: Mapped[str | None] = mapped_column(String)
+    final_name: Mapped[str | None] = mapped_column(String)
+    rationale: Mapped[str | None] = mapped_column(String)
+    confidence: Mapped[float | None] = mapped_column(Float)
+
+    # pending | proposed | edited | approved | error
+    status: Mapped[str] = mapped_column(String, nullable=False, default="pending")
+    error_code: Mapped[str | None] = mapped_column(String)
+    error_message: Mapped[str | None] = mapped_column(String)
+
+    evidence_version: Mapped[str | None] = mapped_column(String)
+    model: Mapped[str | None] = mapped_column(String)
+    prompt_version: Mapped[str | None] = mapped_column(String)
+    raw_response: Mapped[str | None] = mapped_column(String)
+
+    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    approved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )

photo_pipeline/schemas/__init__.py

@@ -1,7 +1,19 @@
 """Pydantic API request/response contracts."""
 
+from photo_pipeline.schemas.albums import (
+    ApproveProposalRequest,
+    EditProposalRequest,
+    GenerateProposalsRequest,
+)
 from photo_pipeline.schemas.duplicates import DecisionRequest
 from photo_pipeline.schemas.jobs import JobStartRequest
 from photo_pipeline.schemas.safety import SafetyDecisionRequest
 
-__all__ = ["DecisionRequest", "JobStartRequest", "SafetyDecisionRequest"]
+__all__ = [
+    "ApproveProposalRequest",
+    "DecisionRequest",
+    "EditProposalRequest",
+    "GenerateProposalsRequest",
+    "JobStartRequest",
+    "SafetyDecisionRequest",
+]

photo_pipeline/schemas/albums.py

@@ -0,0 +1,19 @@
+"""Album proposal API request contracts."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class GenerateProposalsRequest(BaseModel):
+    # None = every album with evidence.
+    albums: list[str] | None = None
+
+
+class EditProposalRequest(BaseModel):
+    name: str
+    expected_version: int
+
+
+class ApproveProposalRequest(BaseModel):
+    expected_version: int

photo_pipeline/services/proposals.py

@@ -0,0 +1,326 @@
+"""ProposalService — AI-assisted, versioned album naming proposals (US03-03).
+
+Generates one durable proposal per album from the US03-01 evidence, names it through
+the US03-02 policy, and keeps rationale, confidence, evidence version, model/prompt
+version, edits, and approval state durable. **No rename ever happens here**: approval
+records intent, and Phase D performs the filesystem work.
+
+Provider minimality — the adapter receives only a compact aggregated summary
+(album/folder name, date, top tags, top locations, counts and a few descriptions).
+Asset IDs, file paths, and raw model responses are never sent, so a proposal cannot
+leak per-photo identity into the naming provider.
+
+Failure handling is explicit and safely retryable: a malformed response, a rate limit,
+or a provider error stores ``status='error'`` with an ``error_code`` and leaves the
+album eligible for another ``generate`` call. Retrying never duplicates a proposal —
+rows are keyed by album, so a regeneration updates the existing row and bumps its
+version.
+
+Staleness: a proposal records the ``evidence_version`` it was generated from. When the
+album's current evidence version differs, the proposal reads back as ``stale`` and
+approval is refused until it is regenerated.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime, timezone
+from typing import Protocol
+
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from photo_pipeline.models import AlbumProposal
+from photo_pipeline.services.albums import AlbumService
+from photo_pipeline.services.naming import NamingPolicy, build_fields, render_name, validate
+
+MODEL = "gemini-2.5-flash"
+PROMPT_VERSION = "1"
+# Bounds on what the provider is allowed to see, so the summary stays minimal.
+MAX_TAGS = 12
+MAX_LOCATIONS = 3
+MAX_DESCRIPTIONS = 5
+
+PENDING = "pending"
+PROPOSED = "proposed"
+EDITED = "edited"
+APPROVED = "approved"
+ERROR = "error"
+
+
+class ProposalProvider(Protocol):
+    def propose(self, summary: dict) -> dict:
+        """Return ``{name, rationale, confidence}`` for one album summary."""
+
+
+class ProposalError(RuntimeError):
+    """Invalid request against a proposal (unknown album, bad name, stale evidence)."""
+
+
+class ConflictError(RuntimeError):
+    """Optimistic version check failed; the proposal changed since it was read."""
+
+
+class ProviderError(RuntimeError):
+    """The naming provider failed. ``code`` distinguishes retryable causes."""
+
+    def __init__(self, code: str, message: str = "") -> None:
+        super().__init__(message or code)
+        self.code = code
+
+
+class RateLimited(ProviderError):
+    def __init__(self, message: str = "rate limited") -> None:
+        super().__init__("rate_limited", message)
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def evidence_summary(evidence: dict) -> dict:
+    """The minimum aggregated evidence a naming provider may see. Bounded and free of
+    asset IDs and file paths."""
+    return {
+        "folder_name": evidence.get("folder_name"),
+        "parent_path": evidence.get("parent_path"),
+        "photo_count": evidence.get("asset_count"),
+        "analyzed_count": evidence.get("analyzed_count"),
+        "dominant_year": evidence.get("dominant_year"),
+        "year_conflict": evidence.get("year_conflict"),
+        "years": [item["value"] for item in (evidence.get("years") or [])],
+        "top_tags": [item["value"] for item in (evidence.get("tags") or [])[:MAX_TAGS]],
+        "locations": [item["value"] for item in (evidence.get("locations") or [])[:MAX_LOCATIONS]],
+        "settings": [item["value"] for item in (evidence.get("settings") or [])],
+        "descriptions": (evidence.get("descriptions") or [])[:MAX_DESCRIPTIONS],
+    }
+
+
+def validate_response(raw: dict) -> dict:
+    """Validate a provider response into ``{name, rationale, confidence}``.
+
+    Raises ``ProviderError('malformed_response')`` for anything unusable, so a bad
+    model reply is quarantined rather than persisted as a name.
+    """
+    if not isinstance(raw, dict):
+        raise ProviderError("malformed_response", "response is not an object")
+    name = raw.get("name")
+    if not isinstance(name, str) or not name.strip():
+        raise ProviderError("malformed_response", "missing or empty 'name'")
+    rationale = raw.get("rationale")
+    if rationale is not None and not isinstance(rationale, str):
+        raise ProviderError("malformed_response", "'rationale' must be a string")
+    confidence = raw.get("confidence")
+    if confidence is not None:
+        if isinstance(confidence, bool) or not isinstance(confidence, (int, float)):
+            raise ProviderError("malformed_response", "'confidence' must be a number")
+        confidence = float(confidence)
+        if not 0.0 <= confidence <= 1.0:
+            raise ProviderError("malformed_response", "'confidence' must be within 0..1")
+    return {"name": name.strip(), "rationale": rationale, "confidence": confidence}
+
+
+class ProposalService:
+    def __init__(
+        self,
+        session_factory: sessionmaker,
+        *,
+        provider: ProposalProvider | None = None,
+        library_roots: tuple = (),
+        policy: NamingPolicy | None = None,
+    ) -> None:
+        self._session_factory = session_factory
+        self._provider = provider
+        self._albums = AlbumService(session_factory, library_roots=library_roots)
+        self._policy = policy or NamingPolicy()
+
+    # ── evidence ──────────────────────────────────────────────────────────────
+
+    def _evidence_by_album(self) -> dict[str, dict]:
+        page = self._albums.aggregate_evidence()
+        return {folder["album"]: folder for folder in page["folders"]}
+
+    # ── generation ────────────────────────────────────────────────────────────
+
+    def generate(self, albums: list[str] | None = None) -> dict:
+        """Generate (or regenerate) proposals. Returns ``{proposed, errors, skipped}``.
+
+        Retry-safe: one row per album, updated in place. A provider failure records an
+        error and leaves the album eligible for another attempt.
+        """
+        evidence_by_album = self._evidence_by_album()
+        targets = albums if albums is not None else sorted(evidence_by_album)
+        proposed = errors = skipped = 0
+
+        for album in targets:
+            evidence = evidence_by_album.get(album)
+            if evidence is None:
+                skipped += 1
+                continue
+            try:
+                result = self._propose_one(evidence)
+            except ProviderError as error:
+                self._store_error(album, error.code, str(error), evidence["version"])
+                errors += 1
+                continue
+            self._store_proposal(album, result, evidence["version"])
+            proposed += 1
+        return {"proposed": proposed, "errors": errors, "skipped": skipped}
+
+    def _propose_one(self, evidence: dict) -> dict:
+        summary = evidence_summary(evidence)
+        if self._provider is None:
+            # No provider configured: fall back to the deterministic policy name, so
+            # proposals still work offline. Rationale says so; confidence stays low.
+            name = render_name(build_fields(evidence), self._policy)
+            if not name:
+                raise ProviderError("no_evidence", "no evidence to build a name from")
+            return {
+                "name": name,
+                "rationale": "Generated from folder evidence using the naming template.",
+                "confidence": 0.3,
+                "raw": None,
+            }
+        raw = self._provider.propose(summary)
+        validated = validate_response(raw)
+        validated["raw"] = json.dumps(raw, ensure_ascii=False, default=str)
+        return validated
+
+    def _store_proposal(self, album: str, result: dict, evidence_version: str) -> None:
+        # Names are always run through the naming policy before they are stored, so a
+        # model can never introduce a path separator or reserved name.
+        checked = validate(result["name"], policy=self._policy)
+        name = checked.name or (checked.suggestions[0] if checked.suggestions else "")
+        with self._session_factory() as session:
+            row = self._row(session, album) or AlbumProposal(id=str(uuid.uuid4()), album=album)
+            row.proposed_name = name
+            row.final_name = None  # a regeneration clears a previous edit
+            row.rationale = result.get("rationale")
+            row.confidence = result.get("confidence")
+            row.status = PROPOSED
+            row.error_code = row.error_message = None
+            row.evidence_version = evidence_version
+            row.model = MODEL if self._provider is not None else "policy"
+            row.prompt_version = PROMPT_VERSION
+            row.raw_response = result.get("raw")
+            row.approved_at = None
+            row.version = (row.version or 0) + 1
+            row.updated_at = _now()
+            session.add(row)
+            session.commit()
+
+    def _store_error(self, album: str, code: str, message: str, evidence_version: str) -> None:
+        with self._session_factory() as session:
+            row = self._row(session, album) or AlbumProposal(id=str(uuid.uuid4()), album=album)
+            row.status = ERROR
+            row.error_code = code
+            row.error_message = message[:500]
+            row.evidence_version = evidence_version
+            row.version = (row.version or 0) + 1
+            row.updated_at = _now()
+            session.add(row)
+            session.commit()
+
+    # ── reads ─────────────────────────────────────────────────────────────────
+
+    @staticmethod
+    def _row(session, album: str) -> AlbumProposal | None:
+        return session.scalar(select(AlbumProposal).where(AlbumProposal.album == album))
+
+    def get(self, album: str) -> dict | None:
+        current = self._evidence_by_album().get(album)
+        with self._session_factory() as session:
+            row = self._row(session, album)
+            if row is None:
+                return None
+            return _as_dict(row, current["version"] if current else None)
+
+    def list(self, *, status: str | None = None) -> dict:
+        versions = {
+            album: evidence["version"] for album, evidence in self._evidence_by_album().items()
+        }
+        with self._session_factory() as session:
+            rows = list(session.scalars(select(AlbumProposal).order_by(AlbumProposal.album)))
+        items = [_as_dict(row, versions.get(row.album)) for row in rows]
+        if status:
+            items = [item for item in items if item["status"] == status]
+        return {"items": items, "total": len(items)}
+
+    # ── edit / approve ────────────────────────────────────────────────────────
+
+    def edit(self, album: str, name: str, *, expected_version: int) -> dict:
+        """Replace the proposed name with the user's own. Validated by the naming
+        policy; an invalid name is rejected with its structured issues."""
+        checked = validate(name, policy=self._policy)
+        if not checked.name:
+            raise ProposalError(f"name is empty after sanitization: {checked.issues}")
+        with self._session_factory() as session:
+            row = self._require(session, album, expected_version)
+            row.final_name = checked.name
+            row.status = EDITED
+            row.approved_at = None
+            row.version += 1
+            row.updated_at = _now()
+            session.add(row)
+            session.commit()
+        return self.get(album)
+
+    def approve(self, album: str, *, expected_version: int) -> dict:
+        """Approve the current proposal. Validates the latest version and that the
+        evidence has not moved on. Records intent only — no rename occurs."""
+        current = self._evidence_by_album().get(album)
+        with self._session_factory() as session:
+            row = self._require(session, album, expected_version)
+            if row.status == ERROR:
+                raise ProposalError(f"proposal for {album!r} is in error state")
+            name = row.final_name or row.proposed_name
+            if not name:
+                raise ProposalError(f"proposal for {album!r} has no name to approve")
+            if current is not None and row.evidence_version != current["version"]:
+                raise ProposalError(
+                    f"proposal for {album!r} is stale; regenerate it before approving"
+                )
+            row.status = APPROVED
+            row.approved_at = _now()
+            row.version += 1
+            row.updated_at = _now()
+            session.add(row)
+            session.commit()
+        return self.get(album)
+
+    def _require(self, session, album: str, expected_version: int) -> AlbumProposal:
+        row = self._row(session, album)
+        if row is None:
+            raise ProposalError(f"unknown album proposal {album!r}")
+        if row.version != expected_version:
+            raise ConflictError(
+                f"proposal for {album!r} is at version {row.version}, expected {expected_version}"
+            )
+        return row
+
+
+def _as_dict(row: AlbumProposal, current_evidence_version: str | None) -> dict:
+    stale = (
+        current_evidence_version is not None
+        and row.evidence_version is not None
+        and row.evidence_version != current_evidence_version
+    )
+    return {
+        "album": row.album,
+        "proposed_name": row.proposed_name,
+        "final_name": row.final_name,
+        "name": row.final_name or row.proposed_name,
+        "rationale": row.rationale,
+        "confidence": row.confidence,
+        "status": row.status,
+        "error_code": row.error_code,
+        "error_message": row.error_message,
+        "evidence_version": row.evidence_version,
+        "current_evidence_version": current_evidence_version,
+        "stale": stale,
+        "model": row.model,
+        "prompt_version": row.prompt_version,
+        "version": row.version,
+        "approved_at": row.approved_at.isoformat() if row.approved_at else None,
+    }

tests/integration/test_album_proposals.py

@@ -0,0 +1,376 @@
+"""Album naming proposals: generation, provider failures, versioning, edits,
+approval, staleness, and durability (US03-03).
+
+Deterministic fakes stand in for the naming provider — success, malformed output,
+rate limiting, and hard failure — so every branch is exercised with no network.
+"""
+
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
+from photo_pipeline.services.proposals import (
+    ConflictError,
+    ProposalError,
+    ProposalService,
+    ProviderError,
+    RateLimited,
+    evidence_summary,
+    validate_response,
+)
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+LIB = (Path("/lib"),)
+
+
+def _factory(tmp_path):
+    (tmp_path / "data").mkdir()
+    config = Config.from_env(
+        {"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"), "PHOTO_PIPELINE_LIBRARY_ROOTS": "/lib"}
+    )
+    run_migrations(config.database_url)
+    return config, create_session_factory(create_db_engine(config.database_url))
+
+
+def _sfw_analyzed(sf, path, **fields):
+    asset_id = str(uuid.uuid4())
+    with sf() as session:
+        session.add(
+            Asset(
+                id=asset_id,
+                original_path=path,
+                current_path=path,
+                discovered_at=NOW,
+                hash_version=1,
+            )
+        )
+        session.add(
+            SafetyReview(id=str(uuid.uuid4()), asset_id=asset_id, decision="sfw", created_at=NOW)
+        )
+        row = AnalysisResult(asset_id=asset_id, status="analyzed")
+        for key, value in fields.items():
+            setattr(row, key, value)
+        session.add(row)
+        session.commit()
+    return asset_id
+
+
+def _seed_rome(sf):
+    _sfw_analyzed(
+        sf,
+        "/lib/rome/a.jpg",
+        description="Colosseum",
+        tags='["ruins"]',
+        approx_year=2019,
+        location_hint="Rome",
+    )
+    _sfw_analyzed(
+        sf,
+        "/lib/rome/b.jpg",
+        description="Forum",
+        tags='["ruins"]',
+        approx_year=2019,
+        location_hint="Rome",
+    )
+
+
+class GoodProvider:
+    """Records what it was shown and returns a valid structured proposal."""
+
+    def __init__(self, name="2019 — Rome — City Trip"):
+        self.seen = []
+        self._name = name
+
+    def propose(self, summary):
+        self.seen.append(summary)
+        return {"name": self._name, "rationale": "Year and place dominate.", "confidence": 0.82}
+
+
+class MalformedProvider:
+    def propose(self, summary):
+        return {"rationale": "no name at all"}
+
+
+class RateLimitedProvider:
+    def __init__(self):
+        self.calls = 0
+
+    def propose(self, summary):
+        self.calls += 1
+        raise RateLimited()
+
+
+class BrokenProvider:
+    def propose(self, summary):
+        raise ProviderError("provider_error", "upstream exploded")
+
+
+def _service(sf, provider=None):
+    return ProposalService(sf, provider=provider, library_roots=LIB)
+
+
+# ── provider contract: minimal evidence, validated response ──────────────────
+
+
+def test_provider_sees_only_minimal_evidence_without_ids_or_paths(tmp_path):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    provider = GoodProvider()
+    _service(sf, provider).generate()
+
+    assert len(provider.seen) == 1
+    summary = provider.seen[0]
+    assert summary["folder_name"] == "rome" and summary["dominant_year"] == 2019
+    assert summary["locations"] == ["Rome"]
+    flat = repr(summary)
+    assert "/lib/" not in flat, "file paths must never reach the naming provider"
+    assert "asset_id" not in summary and "version" not in summary
+
+
+def test_evidence_summary_is_bounded():
+    evidence = {
+        "folder_name": "big",
+        "tags": [{"value": f"t{i}", "count": 1} for i in range(50)],
+        "locations": [{"value": f"l{i}", "count": 1} for i in range(20)],
+        "descriptions": [f"d{i}" for i in range(50)],
+    }
+    summary = evidence_summary(evidence)
+    assert len(summary["top_tags"]) == 12
+    assert len(summary["locations"]) == 3
+    assert len(summary["descriptions"]) == 5
+
+
+@pytest.mark.parametrize(
+    "raw",
+    [
+        "not a dict",
+        {},
+        {"name": ""},
+        {"name": "ok", "confidence": 1.5},
+        {"name": "ok", "confidence": "high"},
+        {"name": "ok", "rationale": 42},
+    ],
+)
+def test_validate_response_rejects_malformed_payloads(raw):
+    with pytest.raises(ProviderError) as caught:
+        validate_response(raw)
+    assert caught.value.code == "malformed_response"
+
+
+def test_validate_response_accepts_a_minimal_valid_payload():
+    assert validate_response({"name": "  Rome  "})["name"] == "Rome"
+
+
+# ── generation, persistence, and failures ────────────────────────────────────
+
+
+def test_generate_persists_proposal_with_rationale_confidence_and_versions(tmp_path):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    service = _service(sf, GoodProvider())
+    assert service.generate() == {"proposed": 1, "errors": 0, "skipped": 0}
+
+    proposal = service.get("rome")
+    assert proposal["name"] == "2019 — Rome — City Trip"
+    assert proposal["rationale"] and proposal["confidence"] == 0.82
+    assert proposal["status"] == "proposed" and proposal["version"] == 1
+    assert proposal["evidence_version"] and proposal["stale"] is False
+    assert proposal["model"] and proposal["prompt_version"]
+
+
+def test_generated_name_is_forced_through_the_naming_policy(tmp_path):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    # A model returning a path separator must never yield a nested name.
+    _service(sf, GoodProvider(name="2019/Rome")).generate()
+    assert _service(sf).get("rome")["name"] == "2019 Rome"
+
+
+@pytest.mark.parametrize(
+    "provider,code",
+    [
+        (MalformedProvider(), "malformed_response"),
+        (RateLimitedProvider(), "rate_limited"),
+        (BrokenProvider(), "provider_error"),
+    ],
+)
+def test_provider_failures_are_explicit_and_retryable(tmp_path, provider, code):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    assert _service(sf, provider).generate() == {"proposed": 0, "errors": 1, "skipped": 0}
+
+    failed = _service(sf).get("rome")
+    assert failed["status"] == "error" and failed["error_code"] == code
+
+    # Retry with a working provider recovers in place — no duplicate row.
+    _service(sf, GoodProvider()).generate()
+    recovered = _service(sf).list()
+    assert recovered["total"] == 1
+    assert recovered["items"][0]["status"] == "proposed"
+    assert recovered["items"][0]["error_code"] is None
+
+
+def test_generate_without_a_provider_falls_back_to_the_policy_name(tmp_path):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    _service(sf).generate()
+    proposal = _service(sf).get("rome")
+    assert proposal["status"] == "proposed" and proposal["model"] == "policy"
+    assert proposal["name"] == "2019 — Rome — rome"
+
+
+def test_generate_skips_albums_without_evidence(tmp_path):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    assert _service(sf, GoodProvider()).generate(["nope"]) == {
+        "proposed": 0,
+        "errors": 0,
+        "skipped": 1,
+    }
+
+
+# ── edit, approve, conflicts, staleness ──────────────────────────────────────
+
+
+def test_edit_sets_the_final_name_and_bumps_the_version(tmp_path):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    service = _service(sf, GoodProvider())
+    service.generate()
+
+    edited = service.edit("rome", "2019 Rome Holiday", expected_version=1)
+    assert edited["final_name"] == "2019 Rome Holiday"
+    assert edited["name"] == "2019 Rome Holiday"
+    assert edited["status"] == "edited" and edited["version"] == 2
+
+
+def test_edit_with_a_stale_version_conflicts_without_mutating(tmp_path):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    service = _service(sf, GoodProvider())
+    service.generate()
+
+    with pytest.raises(ConflictError):
+        service.edit("rome", "Something Else", expected_version=99)
+    assert service.get("rome")["final_name"] is None
+
+
+def test_edit_rejects_a_name_that_sanitizes_to_nothing(tmp_path):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    service = _service(sf, GoodProvider())
+    service.generate()
+    with pytest.raises(ProposalError):
+        service.edit("rome", "///", expected_version=1)
+
+
+def test_approve_validates_the_latest_version_and_renames_nothing(tmp_path):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    service = _service(sf, GoodProvider())
+    service.generate()
+
+    approved = service.approve("rome", expected_version=1)
+    assert approved["status"] == "approved" and approved["approved_at"]
+    # No rename happened: the assets still live at their original paths.
+    with sf() as session:
+        paths = {a.current_path for a in session.scalars(select(Asset))}
+    assert paths == {"/lib/rome/a.jpg", "/lib/rome/b.jpg"}
+
+
+def test_approve_with_a_stale_version_conflicts(tmp_path):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    service = _service(sf, GoodProvider())
+    service.generate()
+    with pytest.raises(ConflictError):
+        service.approve("rome", expected_version=42)
+    assert service.get("rome")["status"] == "proposed"
+
+
+def test_proposal_becomes_stale_when_evidence_changes_and_approval_is_refused(tmp_path):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    service = _service(sf, GoodProvider())
+    service.generate()
+    assert service.get("rome")["stale"] is False
+
+    _sfw_analyzed(sf, "/lib/rome/c.jpg", description="Pantheon", approx_year=2019)
+    stale = service.get("rome")
+    assert stale["stale"] is True
+    with pytest.raises(ProposalError):
+        service.approve("rome", expected_version=stale["version"])
+
+    # Regenerating re-syncs it to the current evidence and clears the staleness.
+    service.generate()
+    assert service.get("rome")["stale"] is False
+
+
+def test_regeneration_clears_a_previous_edit(tmp_path):
+    _, sf = _factory(tmp_path)
+    _seed_rome(sf)
+    service = _service(sf, GoodProvider())
+    service.generate()
+    service.edit("rome", "My Own Name", expected_version=1)
+    service.generate()
+    assert service.get("rome")["final_name"] is None
+
+
+# ── API surface ──────────────────────────────────────────────────────────────
+
+
+def test_api_generate_edit_conflict_approve_and_restart(tmp_path):
+    config, sf = _factory(tmp_path)
+    _seed_rome(sf)
+
+    with TestClient(create_app(config)) as client:
+        assert client.get("/api/v1/albums/evidence").json()["total"] == 1
+
+        generated = client.post("/api/v1/albums/proposals", json={}).json()
+        assert generated["proposed"] == 1
+
+        listed = client.get("/api/v1/albums/proposals").json()
+        assert listed["total"] == 1
+        proposal = listed["items"][0]
+        assert proposal["album"] == "rome" and proposal["status"] == "proposed"
+
+        missing = client.get("/api/v1/albums/proposals/nope")
+        assert missing.status_code == 404
+
+        stale = client.post(
+            "/api/v1/albums/proposals/rome/edit",
+            json={"name": "New Name", "expected_version": 99},
+        )
+        assert stale.status_code == 409
+        assert stale.json()["error"]["code"] == "version_conflict"
+
+        edited = client.post(
+            "/api/v1/albums/proposals/rome/edit",
+            json={"name": "2019 Rome Holiday", "expected_version": proposal["version"]},
+        ).json()
+        assert edited["name"] == "2019 Rome Holiday"
+
+        invalid = client.post(
+            "/api/v1/albums/proposals/rome/edit",
+            json={"name": "///", "expected_version": edited["version"]},
+        )
+        assert invalid.status_code == 422
+
+        approved = client.post(
+            "/api/v1/albums/proposals/rome/approve",
+            json={"expected_version": edited["version"]},
+        ).json()
+        assert approved["status"] == "approved"
+
+    # Restart against the same database: the approval is durable.
+    with TestClient(create_app(config)) as client:
+        after = client.get("/api/v1/albums/proposals/rome").json()
+        assert after["status"] == "approved" and after["name"] == "2019 Rome Holiday"

tests/story_traceability.json

@@ -75,6 +75,9 @@
     ],
     "US03-02": [
       "tests/unit/test_naming_policy.py"
+    ],
+    "US03-03": [
+      "tests/integration/test_album_proposals.py"
     ]
   }
 }