Implement task 014 assets and media library workflow

apps/backend/src/application/articles.py

@@ -64,6 +64,7 @@ def get_article_detail(
     research_manifests = repository.research_manifests.list_for_article(article_id)
     evidence = repository.evidence_items.list_for_article(article_id)
     claims = repository.claims.list_for_article(article_id)
+    assets = repository.assets.list_for_article(article_id)
     agent_jobs = repository.agent_jobs.list_for_article(article_id)
     return ArticleDetailResponse(
         article=article,
@@ -74,6 +75,7 @@ def get_article_detail(
         draft=draft,
         evidence=evidence,
         claims=claims,
+        assets=assets,
         agent_jobs=agent_jobs,
         research_manifests=research_manifests,
     )

apps/backend/src/application/assets.py

@@ -0,0 +1,361 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from pathlib import Path
+from uuid import UUID
+
+from src.domain.contracts import (
+    AssetGenerateSpecsResponse,
+    AssetListResponse,
+    AssetResponse,
+    AssetStatus,
+    AssetSummary,
+    AssetType,
+    AssetUpdateRequest,
+    AssetUploadResponse,
+    CurrentUser,
+    PlanReviewStatus,
+)
+from src.infrastructure.object_storage import ObjectStorageClient
+
+
+def generate_asset_specs(
+    repository: object,
+    *,
+    article_id: UUID,
+    current_user: CurrentUser,
+) -> AssetGenerateSpecsResponse:
+    repository.articles.get(article_id)
+    plan = _approved_plan(repository, article_id=article_id)
+    if plan is None or not plan.sections:
+        raise PermissionError("Approved plan with sections is required before asset specs")
+
+    now = _now()
+    created_assets: list[AssetSummary] = []
+    for index, section in enumerate(plan.sections, start=1):
+        asset_type = _asset_type_for_section(index)
+        title = f"{section.heading} ({asset_type.value})"
+        asset = repository.assets.create(
+            article_id=article_id,
+            section_id=section.id,
+            asset_type=asset_type,
+            title=title,
+            prompt=(
+                "Create an editorial-ready visual for section "
+                f"'{section.heading}' with evidence-backed details."
+            ),
+            object_key=None,
+            file_url=None,
+            alt_text=f"Visual for {section.heading}",
+            caption=f"Supporting visual for section: {section.heading}",
+            status=AssetStatus.GENERATED,
+            created_at=now,
+            updated_at=now,
+        )
+        revision_index = repository.assets.next_revision_index(
+            article_id=article_id,
+            asset_id=asset.id,
+        )
+        repository.assets.create_revision(
+            article_id=article_id,
+            asset_id=asset.id,
+            revision_index=revision_index,
+            action="SPEC_GENERATED",
+            actor_user_id=current_user.id,
+            payload={
+                "section_id": str(section.id),
+                "asset_type": asset.asset_type.value,
+                "title": asset.title,
+            },
+            created_at=now,
+        )
+        created_assets.append(
+            repository.assets.get(article_id=article_id, asset_id=asset.id)
+        )
+
+    repository.articles.create_workflow_event(
+        article_id=article_id,
+        event_type="ASSET_SPECS_GENERATED",
+        from_status=None,
+        to_status=None,
+        actor_user_id=current_user.id,
+        payload={
+            "count": len(created_assets),
+            "asset_ids": [str(asset.id) for asset in created_assets],
+        },
+        created_at=now,
+    )
+    return AssetGenerateSpecsResponse(assets=created_assets)
+
+
+def list_assets(repository: object, *, article_id: UUID) -> AssetListResponse:
+    repository.articles.get(article_id)
+    return AssetListResponse(assets=repository.assets.list_for_article(article_id))
+
+
+def update_asset(
+    repository: object,
+    *,
+    article_id: UUID,
+    asset_id: UUID,
+    request: AssetUpdateRequest,
+    current_user: CurrentUser,
+) -> AssetResponse:
+    current = repository.assets.get(article_id=article_id, asset_id=asset_id)
+    next_section_id = request.section_id if request.section_id is not None else current.section_id
+    next_title = request.title if request.title is not None else current.title
+    next_prompt = request.prompt if request.prompt is not None else current.prompt
+    next_alt_text = request.alt_text if request.alt_text is not None else current.alt_text
+    next_caption = request.caption if request.caption is not None else current.caption
+    next_status = request.status if request.status is not None else current.status
+    changed_fields = _changed_fields(
+        current=current,
+        section_id=next_section_id,
+        title=next_title,
+        prompt=next_prompt,
+        alt_text=next_alt_text,
+        caption=next_caption,
+        status=next_status,
+    )
+    if not changed_fields:
+        return AssetResponse(asset=current)
+
+    now = _now()
+    repository.assets.update(
+        article_id=article_id,
+        asset_id=asset_id,
+        section_id=next_section_id,
+        title=next_title,
+        prompt=next_prompt,
+        object_key=current.object_key,
+        file_url=current.file_url,
+        alt_text=next_alt_text,
+        caption=next_caption,
+        status=next_status,
+        updated_at=now,
+    )
+    revision_index = repository.assets.next_revision_index(
+        article_id=article_id,
+        asset_id=asset_id,
+    )
+    repository.assets.create_revision(
+        article_id=article_id,
+        asset_id=asset_id,
+        revision_index=revision_index,
+        action="ASSET_UPDATED",
+        actor_user_id=current_user.id,
+        payload={
+            "changed_fields": changed_fields,
+            "status": next_status.value,
+        },
+        created_at=now,
+    )
+    if current.status != next_status:
+        repository.articles.create_workflow_event(
+            article_id=article_id,
+            event_type="ASSET_STATUS_UPDATED",
+            from_status=None,
+            to_status=None,
+            actor_user_id=current_user.id,
+            payload={
+                "asset_id": str(asset_id),
+                "from_status": current.status.value,
+                "to_status": next_status.value,
+            },
+            created_at=now,
+        )
+    refreshed = repository.assets.get(article_id=article_id, asset_id=asset_id)
+    return AssetResponse(asset=refreshed)
+
+
+def approve_asset(
+    repository: object,
+    *,
+    article_id: UUID,
+    asset_id: UUID,
+    current_user: CurrentUser,
+) -> AssetResponse:
+    current = repository.assets.get(article_id=article_id, asset_id=asset_id)
+    if current.status == AssetStatus.APPROVED:
+        return AssetResponse(asset=current)
+
+    now = _now()
+    repository.assets.update(
+        article_id=article_id,
+        asset_id=asset_id,
+        section_id=current.section_id,
+        title=current.title,
+        prompt=current.prompt,
+        object_key=current.object_key,
+        file_url=current.file_url,
+        alt_text=current.alt_text,
+        caption=current.caption,
+        status=AssetStatus.APPROVED,
+        updated_at=now,
+    )
+    approved = repository.assets.get(article_id=article_id, asset_id=asset_id)
+    revision_index = repository.assets.next_revision_index(
+        article_id=article_id,
+        asset_id=asset_id,
+    )
+    repository.assets.create_revision(
+        article_id=article_id,
+        asset_id=asset_id,
+        revision_index=revision_index,
+        action="ASSET_APPROVED",
+        actor_user_id=current_user.id,
+        payload={
+            "from_status": current.status.value,
+            "to_status": AssetStatus.APPROVED.value,
+            "object_key": approved.object_key,
+        },
+        created_at=now,
+    )
+    repository.articles.create_workflow_event(
+        article_id=article_id,
+        event_type="ASSET_APPROVED",
+        from_status=None,
+        to_status=None,
+        actor_user_id=current_user.id,
+        payload={
+            "asset_id": str(asset_id),
+            "from_status": current.status.value,
+            "to_status": AssetStatus.APPROVED.value,
+            "object_key": approved.object_key,
+            "file_url": approved.file_url,
+        },
+        created_at=now,
+    )
+    refreshed = repository.assets.get(article_id=article_id, asset_id=asset_id)
+    return AssetResponse(asset=refreshed)
+
+
+def upload_asset_file(
+    repository: object,
+    *,
+    article_id: UUID,
+    asset_id: UUID,
+    filename: str,
+    content_type: str | None,
+    content: bytes,
+    current_user: CurrentUser,
+    object_storage: ObjectStorageClient,
+) -> AssetUploadResponse:
+    current = repository.assets.get(article_id=article_id, asset_id=asset_id)
+    now = _now()
+    revision_index = repository.assets.next_revision_index(
+        article_id=article_id,
+        asset_id=asset_id,
+    )
+    safe_filename = _safe_filename(filename)
+    object_key = f"articles/{article_id}/assets/{asset_id}/r{revision_index}/{safe_filename}"
+    file_url = object_storage.put_bytes(
+        object_key=object_key,
+        content=content,
+        content_type=content_type or "application/octet-stream",
+    )
+    had_previous_file = bool(current.object_key or current.file_url)
+    repository.assets.update(
+        article_id=article_id,
+        asset_id=asset_id,
+        section_id=current.section_id,
+        title=current.title,
+        prompt=current.prompt,
+        object_key=object_key,
+        file_url=file_url,
+        alt_text=current.alt_text,
+        caption=current.caption,
+        status=AssetStatus.PENDING,
+        updated_at=now,
+    )
+    action = "ASSET_FILE_REPLACED" if had_previous_file else "ASSET_FILE_UPLOADED"
+    repository.assets.create_revision(
+        article_id=article_id,
+        asset_id=asset_id,
+        revision_index=revision_index,
+        action=action,
+        actor_user_id=current_user.id,
+        payload={
+            "previous_object_key": current.object_key,
+            "previous_file_url": current.file_url,
+            "next_object_key": object_key,
+            "next_file_url": file_url,
+            "content_type": content_type or "application/octet-stream",
+            "filename": safe_filename,
+        },
+        created_at=now,
+    )
+    repository.articles.create_workflow_event(
+        article_id=article_id,
+        event_type=action,
+        from_status=None,
+        to_status=None,
+        actor_user_id=current_user.id,
+        payload={
+            "asset_id": str(asset_id),
+            "previous_object_key": current.object_key,
+            "next_object_key": object_key,
+        },
+        created_at=now,
+    )
+    refreshed = repository.assets.get(article_id=article_id, asset_id=asset_id)
+    return AssetUploadResponse(asset=refreshed)
+
+
+def _approved_plan(repository: object, *, article_id: UUID) -> object | None:
+    plans = repository.article_plans.list_for_article(article_id)
+    approved_plans = [plan for plan in plans if plan.status == PlanReviewStatus.APPROVED]
+    if not approved_plans:
+        return None
+    approved_plans.sort(key=lambda plan: plan.version)
+    return approved_plans[-1]
+
+
+def _asset_type_for_section(index: int) -> AssetType:
+    ordered_types = [
+        AssetType.HERO_IMAGE,
+        AssetType.ARCHITECTURE_DIAGRAM,
+        AssetType.DIAGRAM,
+        AssetType.TABLE,
+        AssetType.FLOWCHART,
+        AssetType.COMPARISON_MATRIX,
+        AssetType.INLINE_IMAGE,
+    ]
+    return ordered_types[(index - 1) % len(ordered_types)]
+
+
+def _changed_fields(
+    *,
+    current: AssetSummary,
+    section_id: UUID | None,
+    title: str,
+    prompt: str | None,
+    alt_text: str | None,
+    caption: str | None,
+    status: AssetStatus,
+) -> list[str]:
+    changed: list[str] = []
+    if current.section_id != section_id:
+        changed.append("section_id")
+    if current.title != title:
+        changed.append("title")
+    if current.prompt != prompt:
+        changed.append("prompt")
+    if current.alt_text != alt_text:
+        changed.append("alt_text")
+    if current.caption != caption:
+        changed.append("caption")
+    if current.status != status:
+        changed.append("status")
+    return changed
+
+
+def _safe_filename(value: str) -> str:
+    filename = Path(value).name.strip()
+    if not filename:
+        return "asset.bin"
+    return filename.replace("/", "_")
+
+
+def _now() -> datetime:
+    return datetime.now(UTC)

apps/backend/src/application/drafts.py

@@ -9,6 +9,7 @@ from src.domain.contracts import (
     AgentJobStatus,
     AgentJobSummary,
     AgentJobType,
+    AssetStatus,
     ArticleWorkflowStatus,
     DraftAssembleResponse,
     DraftFaqItem,
@@ -51,6 +52,10 @@ def assemble_draft(repository: object, *, article_id: UUID) -> DraftAssembleResp
     visual_placeholders: list[str] = []
     evidence_references: list[str] = []
     unsupported_warnings = _collect_job_warnings(section_jobs)
+    approved_assets_by_section = _approved_assets_by_section(
+        repository,
+        article_id=article_id,
+    )
 
     for section in approved_plan.sections:
         section_job = successful_jobs_by_section[str(section.id)]
@@ -61,6 +66,7 @@ def assemble_draft(repository: object, *, article_id: UUID) -> DraftAssembleResp
                 section_id=section.id,
                 heading=section.heading,
                 payload=payload,
+                approved_assets=approved_assets_by_section.get(str(section.id), []),
             )
         )
         evidence_references.extend(_string_list(payload.get("used_evidence_ids")))
@@ -279,7 +285,11 @@ def _visual_placeholders_for_section(
     section_id: UUID,
     heading: str,
     payload: dict[str, Any],
+    approved_assets: list[object],
 ) -> list[str]:
+    if approved_assets:
+        return [_approved_asset_markdown(asset) for asset in approved_assets]
+
     suggested_visuals = payload.get("suggested_visuals")
     placeholders: list[str] = []
     if isinstance(suggested_visuals, list):
@@ -293,6 +303,30 @@ def _visual_placeholders_for_section(
     return [f"![Visual placeholder: {heading}](placeholder://{section_id}/1)"]
 
 
+def _approved_assets_by_section(repository: object, *, article_id: UUID) -> dict[str, list[object]]:
+    assets = repository.assets.list_for_article(article_id)
+    by_section: dict[str, list[object]] = {}
+    for asset in assets:
+        if asset.status != AssetStatus.APPROVED:
+            continue
+        if not asset.section_id:
+            continue
+        if not asset.file_url:
+            continue
+        by_section.setdefault(str(asset.section_id), []).append(asset)
+    return by_section
+
+
+def _approved_asset_markdown(asset: object) -> str:
+    alt = _string_value(getattr(asset, "alt_text", None)) or _string_value(
+        getattr(asset, "title", None)
+    )
+    src = _string_value(getattr(asset, "file_url", None))
+    if not src:
+        return ""
+    return f"![{alt}]({src})"
+
+
 def _visual_label(value: Any, *, heading: str, fallback_index: int) -> str:
     if isinstance(value, str) and value.strip():
         return value.strip()

apps/backend/src/domain/contracts/__init__.py

@@ -30,7 +30,14 @@ from .models import (
     BoundaryQuestionSummary,
     BoundaryQuestionUpdateRequest,
     WorkflowEventSummary,
+    AssetGenerateSpecsResponse,
+    AssetListResponse,
+    AssetResponse,
+    AssetRevisionSummary,
     AssetSummary,
+    AssetUpdateRequest,
+    AssetUploadRequest,
+    AssetUploadResponse,
     ClaimSummary,
     CurrentUser,
     CurrentUserResponse,
@@ -98,10 +105,17 @@ __all__ = [
     "BoundaryQuestionSummary",
     "BoundaryQuestionUpdateRequest",
     "WorkflowEventSummary",
+    "AssetGenerateSpecsResponse",
+    "AssetListResponse",
+    "AssetResponse",
+    "AssetRevisionSummary",
     "ArticleWorkflowStatus",
     "AssetStatus",
     "AssetSummary",
     "AssetType",
+    "AssetUpdateRequest",
+    "AssetUploadRequest",
+    "AssetUploadResponse",
     "ClaimRiskLevel",
     "ClaimSummary",
     "ClaimSupportStatus",

apps/backend/src/domain/contracts/enums.py

@@ -97,6 +97,9 @@ class AssetType(str, Enum):
     HERO_IMAGE = "hero_image"
     DIAGRAM = "diagram"
     TABLE = "table"
+    FLOWCHART = "flowchart"
+    COMPARISON_MATRIX = "comparison_matrix"
+    ARCHITECTURE_DIAGRAM = "architecture_diagram"
     INLINE_IMAGE = "inline_image"
 
 

apps/backend/src/domain/contracts/models.py

@@ -386,16 +386,32 @@ class DraftSummary(ContractModel):
     updated_at: datetime
 
 
+class AssetRevisionSummary(ContractModel):
+    id: UUID
+    asset_id: UUID
+    article_id: UUID
+    revision_index: int = Field(ge=1)
+    action: str = Field(min_length=1)
+    actor_user_id: UUID | None = None
+    payload: JsonObject = Field(default_factory=dict)
+    created_at: datetime
+
+
 class AssetSummary(ContractModel):
     id: UUID
     article_id: UUID
+    section_id: UUID | None = None
     asset_type: AssetType
     title: str = Field(min_length=1)
     prompt: str | None = None
+    object_key: str | None = None
     file_url: str | None = None
     alt_text: str | None = None
     caption: str | None = None
     status: AssetStatus
+    history: list[AssetRevisionSummary] = Field(default_factory=list)
+    created_at: datetime
+    updated_at: datetime
 
 
 class ReviewSummary(ContractModel):
@@ -497,6 +513,37 @@ class DraftUpdateRequest(ContractModel):
     unsupported_claim_warnings: list[str] | None = None
 
 
+class AssetGenerateSpecsResponse(ContractModel):
+    assets: list[AssetSummary] = Field(default_factory=list)
+
+
+class AssetListResponse(ContractModel):
+    assets: list[AssetSummary] = Field(default_factory=list)
+
+
+class AssetResponse(ContractModel):
+    asset: AssetSummary
+
+
+class AssetUpdateRequest(ContractModel):
+    title: str | None = Field(default=None, min_length=1)
+    prompt: str | None = None
+    alt_text: str | None = None
+    caption: str | None = None
+    section_id: UUID | None = None
+    status: AssetStatus | None = None
+
+
+class AssetUploadRequest(ContractModel):
+    filename: str = Field(min_length=1)
+    content_base64: str = Field(min_length=1)
+    content_type: str | None = None
+
+
+class AssetUploadResponse(ContractModel):
+    asset: AssetSummary
+
+
 class ArticleDetailResponse(ContractModel):
     article: ArticleSummary
     target_site: TargetSiteConfig | None = None

apps/backend/src/domain/contracts/openapi.py

@@ -36,7 +36,14 @@ from .models import (
     BoundaryQuestionResponse,
     BoundaryQuestionSummary,
     BoundaryQuestionUpdateRequest,
+    AssetGenerateSpecsResponse,
+    AssetListResponse,
+    AssetResponse,
+    AssetRevisionSummary,
     AssetSummary,
+    AssetUpdateRequest,
+    AssetUploadRequest,
+    AssetUploadResponse,
     ClaimSummary,
     CurrentUser,
     CurrentUserResponse,
@@ -142,7 +149,14 @@ CONTRACT_SCHEMA_MODELS: tuple[type[BaseModel], ...] = (
     DraftListResponse,
     DraftResponse,
     DraftUpdateRequest,
+    AssetGenerateSpecsResponse,
+    AssetListResponse,
+    AssetResponse,
+    AssetRevisionSummary,
     AssetSummary,
+    AssetUpdateRequest,
+    AssetUploadRequest,
+    AssetUploadResponse,
     ReviewSummary,
     ReviewActionResponse,
     PublishCommitSummary,

apps/backend/src/domain/schema.py

@@ -12,6 +12,7 @@ EVIDENCE_ITEMS_TABLE = "evidence_items"
 CLAIMS_TABLE = "claims"
 ARTICLE_DRAFTS_TABLE = "article_drafts"
 ASSETS_TABLE = "assets"
+ASSET_REVISIONS_TABLE = "asset_revisions"
 WORKFLOW_EVENTS_TABLE = "workflow_events"
 AGENT_JOBS_TABLE = "agent_jobs"
 RESEARCH_RUN_MANIFESTS_TABLE = "research_run_manifests"
@@ -31,6 +32,7 @@ CORE_TABLES: tuple[str, ...] = (
     CLAIMS_TABLE,
     ARTICLE_DRAFTS_TABLE,
     ASSETS_TABLE,
+    ASSET_REVISIONS_TABLE,
     WORKFLOW_EVENTS_TABLE,
     AGENT_JOBS_TABLE,
     RESEARCH_RUN_MANIFESTS_TABLE,

apps/backend/src/infrastructure/object_storage.py

@@ -8,6 +8,15 @@ class ObjectStorageClient:
     def put_text(self, *, object_key: str, content: str) -> str:
         raise NotImplementedError
 
+    def put_bytes(
+        self,
+        *,
+        object_key: str,
+        content: bytes,
+        content_type: str = "application/octet-stream",
+    ) -> str:
+        raise NotImplementedError
+
 
 class LocalObjectStorageClient(ObjectStorageClient):
     def __init__(self, root: Path) -> None:
@@ -19,6 +28,19 @@ class LocalObjectStorageClient(ObjectStorageClient):
         path.write_text(content, encoding="utf-8")
         return f"file://{path}"
 
+    def put_bytes(
+        self,
+        *,
+        object_key: str,
+        content: bytes,
+        content_type: str = "application/octet-stream",
+    ) -> str:
+        del content_type
+        path = self.root / object_key
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(content)
+        return f"file://{path}"
+
 
 class S3ObjectStorageClient(ObjectStorageClient):
     def __init__(self) -> None:
@@ -36,11 +58,24 @@ class S3ObjectStorageClient(ObjectStorageClient):
         )
 
     def put_text(self, *, object_key: str, content: str) -> str:
+        return self.put_bytes(
+            object_key=object_key,
+            content=content.encode("utf-8"),
+            content_type="application/json; charset=utf-8",
+        )
+
+    def put_bytes(
+        self,
+        *,
+        object_key: str,
+        content: bytes,
+        content_type: str = "application/octet-stream",
+    ) -> str:
         self.client.put_object(
             Bucket=self.bucket,
             Key=object_key,
-            Body=content.encode("utf-8"),
-            ContentType="application/json; charset=utf-8",
+            Body=content,
+            ContentType=content_type,
         )
         return f"s3://{self.bucket}/{object_key}"
 

apps/backend/src/infrastructure/repositories.py

@@ -13,6 +13,10 @@ from src.domain.contracts import (
     AgentJobStatus,
     AgentJobSummary,
     AgentJobType,
+    AssetRevisionSummary,
+    AssetStatus,
+    AssetSummary,
+    AssetType,
     ArticleSummary,
     BoundaryQuestionSummary,
     ClaimRiskLevel,
@@ -55,6 +59,7 @@ class BackendRepository:
         self.boundary_questions = BoundaryQuestionsRepository(self)
         self.article_plans = ArticlePlansRepository(self)
         self.article_drafts = ArticleDraftsRepository(self)
+        self.assets = AssetsRepository(self)
         self.research_manifests = ResearchManifestsRepository(self)
         self.evidence_items = EvidenceItemsRepository(self)
         self.claims = ClaimsRepository(self)
@@ -1192,6 +1197,325 @@ class ArticleDraftsRepository:
         """
 
 
+class AssetsRepository:
+    def __init__(self, repository: BackendRepository) -> None:
+        self._repository = repository
+
+    def create(
+        self,
+        *,
+        article_id: UUID,
+        section_id: UUID | None,
+        asset_type: AssetType,
+        title: str,
+        prompt: str | None,
+        object_key: str | None,
+        file_url: str | None,
+        alt_text: str | None,
+        caption: str | None,
+        status: AssetStatus,
+        created_at: datetime,
+        updated_at: datetime,
+    ) -> AssetSummary:
+        asset_id = uuid4()
+        placeholder = self._repository.placeholder()
+        with self._repository.connection() as connection:
+            connection.execute(
+                f"""
+                INSERT INTO assets (
+                    id,
+                    article_id,
+                    section_id,
+                    asset_type,
+                    title,
+                    prompt,
+                    object_key,
+                    file_url,
+                    alt_text,
+                    caption,
+                    status,
+                    created_at,
+                    updated_at
+                )
+                VALUES (
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder}
+                )
+                """,
+                (
+                    str(asset_id),
+                    str(article_id),
+                    _uuid_value(section_id),
+                    asset_type.value,
+                    title,
+                    prompt,
+                    object_key,
+                    file_url,
+                    alt_text,
+                    caption,
+                    status.value,
+                    _datetime_value(created_at),
+                    _datetime_value(updated_at),
+                ),
+            )
+
+        return self.get(article_id=article_id, asset_id=asset_id)
+
+    def list_for_article(self, article_id: UUID) -> list[AssetSummary]:
+        placeholder = self._repository.placeholder()
+        with self._repository.connection() as connection:
+            rows = connection.execute(
+                f"""
+                SELECT
+                    id,
+                    article_id,
+                    section_id,
+                    asset_type,
+                    title,
+                    prompt,
+                    object_key,
+                    file_url,
+                    alt_text,
+                    caption,
+                    status,
+                    created_at,
+                    updated_at
+                FROM assets
+                WHERE article_id = {placeholder}
+                ORDER BY created_at, id
+                """,
+                (str(article_id),),
+            ).fetchall()
+
+        history_by_asset = self._history_by_asset(article_id)
+        assets: list[AssetSummary] = []
+        for row in rows:
+            asset = _asset_summary_from_row(
+                row,
+                history=history_by_asset.get(str(_row_value(row, "id")), []),
+            )
+            assets.append(asset)
+        return assets
+
+    def get(self, *, article_id: UUID, asset_id: UUID) -> AssetSummary:
+        placeholder = self._repository.placeholder()
+        with self._repository.connection() as connection:
+            row = connection.execute(
+                f"""
+                SELECT
+                    id,
+                    article_id,
+                    section_id,
+                    asset_type,
+                    title,
+                    prompt,
+                    object_key,
+                    file_url,
+                    alt_text,
+                    caption,
+                    status,
+                    created_at,
+                    updated_at
+                FROM assets
+                WHERE article_id = {placeholder} AND id = {placeholder}
+                """,
+                (str(article_id), str(asset_id)),
+            ).fetchone()
+        if row is None:
+            raise LookupError(f"Asset not found: {asset_id}")
+        history_by_asset = self._history_by_asset(article_id)
+        return _asset_summary_from_row(
+            row,
+            history=history_by_asset.get(str(asset_id), []),
+        )
+
+    def update(
+        self,
+        *,
+        article_id: UUID,
+        asset_id: UUID,
+        section_id: UUID | None,
+        title: str,
+        prompt: str | None,
+        object_key: str | None,
+        file_url: str | None,
+        alt_text: str | None,
+        caption: str | None,
+        status: AssetStatus,
+        updated_at: datetime,
+    ) -> AssetSummary:
+        placeholder = self._repository.placeholder()
+        with self._repository.connection() as connection:
+            connection.execute(
+                f"""
+                UPDATE assets
+                SET
+                    section_id = {placeholder},
+                    title = {placeholder},
+                    prompt = {placeholder},
+                    object_key = {placeholder},
+                    file_url = {placeholder},
+                    alt_text = {placeholder},
+                    caption = {placeholder},
+                    status = {placeholder},
+                    updated_at = {placeholder}
+                WHERE article_id = {placeholder} AND id = {placeholder}
+                """,
+                (
+                    _uuid_value(section_id),
+                    title,
+                    prompt,
+                    object_key,
+                    file_url,
+                    alt_text,
+                    caption,
+                    status.value,
+                    _datetime_value(updated_at),
+                    str(article_id),
+                    str(asset_id),
+                ),
+            )
+
+        return self.get(article_id=article_id, asset_id=asset_id)
+
+    def next_revision_index(self, *, article_id: UUID, asset_id: UUID) -> int:
+        placeholder = self._repository.placeholder()
+        with self._repository.connection() as connection:
+            row = connection.execute(
+                f"""
+                SELECT COALESCE(MAX(revision_index), 0) + 1 AS next_revision
+                FROM asset_revisions
+                WHERE article_id = {placeholder} AND asset_id = {placeholder}
+                """,
+                (str(article_id), str(asset_id)),
+            ).fetchone()
+        return int(_row_value(row, "next_revision"))
+
+    def create_revision(
+        self,
+        *,
+        article_id: UUID,
+        asset_id: UUID,
+        revision_index: int,
+        action: str,
+        actor_user_id: UUID | None,
+        payload: JsonObject,
+        created_at: datetime,
+    ) -> AssetRevisionSummary:
+        revision_id = uuid4()
+        placeholder = self._repository.placeholder()
+        json_cast = self._repository.json_cast()
+        with self._repository.connection() as connection:
+            connection.execute(
+                f"""
+                INSERT INTO asset_revisions (
+                    id,
+                    article_id,
+                    asset_id,
+                    revision_index,
+                    action,
+                    actor_user_id,
+                    payload,
+                    created_at
+                )
+                VALUES (
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder},
+                    {placeholder}{json_cast},
+                    {placeholder}
+                )
+                """,
+                (
+                    str(revision_id),
+                    str(article_id),
+                    str(asset_id),
+                    revision_index,
+                    action,
+                    _uuid_value(actor_user_id),
+                    _json_value(payload),
+                    _datetime_value(created_at),
+                ),
+            )
+
+        return self.get_revision(
+            article_id=article_id,
+            asset_id=asset_id,
+            revision_id=revision_id,
+        )
+
+    def get_revision(
+        self,
+        *,
+        article_id: UUID,
+        asset_id: UUID,
+        revision_id: UUID,
+    ) -> AssetRevisionSummary:
+        placeholder = self._repository.placeholder()
+        with self._repository.connection() as connection:
+            row = connection.execute(
+                f"""
+                SELECT
+                    id,
+                    article_id,
+                    asset_id,
+                    revision_index,
+                    action,
+                    actor_user_id,
+                    payload,
+                    created_at
+                FROM asset_revisions
+                WHERE article_id = {placeholder}
+                  AND asset_id = {placeholder}
+                  AND id = {placeholder}
+                """,
+                (str(article_id), str(asset_id), str(revision_id)),
+            ).fetchone()
+        if row is None:
+            raise LookupError(f"Asset revision not found: {revision_id}")
+        return _asset_revision_from_row(row)
+
+    def _history_by_asset(self, article_id: UUID) -> dict[str, list[AssetRevisionSummary]]:
+        placeholder = self._repository.placeholder()
+        with self._repository.connection() as connection:
+            rows = connection.execute(
+                f"""
+                SELECT
+                    id,
+                    article_id,
+                    asset_id,
+                    revision_index,
+                    action,
+                    actor_user_id,
+                    payload,
+                    created_at
+                FROM asset_revisions
+                WHERE article_id = {placeholder}
+                ORDER BY revision_index, created_at, id
+                """,
+                (str(article_id),),
+            ).fetchall()
+        history: dict[str, list[AssetRevisionSummary]] = {}
+        for row in rows:
+            asset_id = str(_row_value(row, "asset_id"))
+            history.setdefault(asset_id, []).append(_asset_revision_from_row(row))
+        return history
+
+
 class ResearchManifestsRepository:
     def __init__(self, repository: BackendRepository) -> None:
         self._repository = repository
@@ -2383,6 +2707,42 @@ def _draft_summary_from_row(row: Any) -> DraftSummary:
     )
 
 
+def _asset_revision_from_row(row: Any) -> AssetRevisionSummary:
+    return AssetRevisionSummary(
+        id=_row_value(row, "id"),
+        article_id=_row_value(row, "article_id"),
+        asset_id=_row_value(row, "asset_id"),
+        revision_index=_row_value(row, "revision_index"),
+        action=_row_value(row, "action"),
+        actor_user_id=_row_value(row, "actor_user_id"),
+        payload=_json_from_row(row, "payload"),
+        created_at=_row_value(row, "created_at"),
+    )
+
+
+def _asset_summary_from_row(
+    row: Any,
+    *,
+    history: list[AssetRevisionSummary] | None = None,
+) -> AssetSummary:
+    return AssetSummary(
+        id=_row_value(row, "id"),
+        article_id=_row_value(row, "article_id"),
+        section_id=_row_value(row, "section_id"),
+        asset_type=_row_value(row, "asset_type"),
+        title=_row_value(row, "title"),
+        prompt=_row_value(row, "prompt"),
+        object_key=_row_value(row, "object_key"),
+        file_url=_row_value(row, "file_url"),
+        alt_text=_row_value(row, "alt_text"),
+        caption=_row_value(row, "caption"),
+        status=_row_value(row, "status"),
+        history=history or [],
+        created_at=_row_value(row, "created_at"),
+        updated_at=_row_value(row, "updated_at"),
+    )
+
+
 def _agent_job_summary_from_row(row: Any) -> AgentJobSummary:
     return AgentJobSummary(
         id=_row_value(row, "id"),

apps/backend/src/infrastructure/schema.py

@@ -237,9 +237,11 @@ POSTGRES_SCHEMA_STATEMENTS: tuple[str, ...] = (
     CREATE TABLE IF NOT EXISTS assets (
         id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
         article_id UUID NOT NULL REFERENCES articles(id) ON DELETE CASCADE,
+        section_id UUID REFERENCES plan_sections(id) ON DELETE SET NULL,
         asset_type TEXT NOT NULL CHECK (asset_type IN ({ASSET_TYPE_VALUES})),
         title TEXT NOT NULL,
         prompt TEXT,
+        object_key TEXT,
         file_url TEXT,
         alt_text TEXT,
         caption TEXT,
@@ -248,6 +250,20 @@ POSTGRES_SCHEMA_STATEMENTS: tuple[str, ...] = (
         updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
     )
     """,
+    "ALTER TABLE assets ADD COLUMN IF NOT EXISTS section_id UUID REFERENCES plan_sections(id) ON DELETE SET NULL",
+    "ALTER TABLE assets ADD COLUMN IF NOT EXISTS object_key TEXT",
+    f"""
+    CREATE TABLE IF NOT EXISTS asset_revisions (
+        id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+        article_id UUID NOT NULL REFERENCES articles(id) ON DELETE CASCADE,
+        asset_id UUID NOT NULL REFERENCES assets(id) ON DELETE CASCADE,
+        revision_index INTEGER NOT NULL,
+        action TEXT NOT NULL,
+        actor_user_id UUID REFERENCES users(id),
+        payload JSONB NOT NULL DEFAULT '{{}}'::jsonb,
+        created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+    )
+    """,
     f"""
     CREATE TABLE IF NOT EXISTS workflow_events (
         id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
@@ -370,6 +386,10 @@ POSTGRES_SCHEMA_STATEMENTS: tuple[str, ...] = (
     ON workflow_events (article_id, created_at)
     """,
     """
+    CREATE INDEX IF NOT EXISTS idx_asset_revisions_asset
+    ON asset_revisions (asset_id, revision_index, created_at)
+    """,
+    """
     CREATE INDEX IF NOT EXISTS idx_research_run_manifests_article
     ON research_run_manifests (article_id, created_at)
     """,
@@ -553,9 +573,11 @@ SQLITE_SCHEMA_STATEMENTS: tuple[str, ...] = (
     CREATE TABLE IF NOT EXISTS assets (
         id TEXT PRIMARY KEY,
         article_id TEXT NOT NULL,
+        section_id TEXT,
         asset_type TEXT NOT NULL,
         title TEXT NOT NULL,
         prompt TEXT,
+        object_key TEXT,
         file_url TEXT,
         alt_text TEXT,
         caption TEXT,
@@ -565,6 +587,18 @@ SQLITE_SCHEMA_STATEMENTS: tuple[str, ...] = (
     )
     """,
     """
+    CREATE TABLE IF NOT EXISTS asset_revisions (
+        id TEXT PRIMARY KEY,
+        article_id TEXT NOT NULL,
+        asset_id TEXT NOT NULL,
+        revision_index INTEGER NOT NULL,
+        action TEXT NOT NULL,
+        actor_user_id TEXT,
+        payload TEXT NOT NULL DEFAULT '{}',
+        created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+    )
+    """,
+    """
     CREATE TABLE IF NOT EXISTS workflow_events (
         id TEXT PRIMARY KEY,
         article_id TEXT NOT NULL,
@@ -676,6 +710,10 @@ SQLITE_SCHEMA_STATEMENTS: tuple[str, ...] = (
     ON workflow_events (article_id, created_at)
     """,
     """
+    CREATE INDEX IF NOT EXISTS idx_asset_revisions_asset
+    ON asset_revisions (asset_id, revision_index, created_at)
+    """,
+    """
     CREATE INDEX IF NOT EXISTS idx_research_run_manifests_article
     ON research_run_manifests (article_id, created_at)
     """,

apps/backend/src/presentation/main.py

@@ -10,6 +10,7 @@ from src.presentation.routes.agent_jobs import (
 )
 from src.presentation.routes.agent_jobs import router as agent_jobs_router
 from src.presentation.routes.articles import router as articles_router
+from src.presentation.routes.assets import router as assets_router
 from src.presentation.routes.auth import router as auth_router
 from src.presentation.routes.boundary_questions import router as boundary_questions_router
 from src.presentation.routes.drafts import router as drafts_router
@@ -21,6 +22,7 @@ from src.presentation.routes.sites import router as sites_router
 app = FastAPI(title="AI Content Pipeline Backend")
 app.include_router(auth_router)
 app.include_router(articles_router)
+app.include_router(assets_router)
 app.include_router(boundary_questions_router)
 app.include_router(plans_router)
 app.include_router(evidence_router)

apps/backend/src/presentation/routes/assets.py

@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+import base64
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, status
+
+from src.application.assets import (
+    approve_asset,
+    generate_asset_specs,
+    list_assets,
+    update_asset,
+    upload_asset_file,
+)
+from src.domain.auth import EDITOR_OR_ADMIN_ROLES
+from src.domain.contracts import (
+    AssetGenerateSpecsResponse,
+    AssetListResponse,
+    AssetResponse,
+    AssetUpdateRequest,
+    AssetUploadRequest,
+    AssetUploadResponse,
+    CurrentUser,
+)
+from src.infrastructure.object_storage import open_object_storage_client
+from src.infrastructure.repositories import BackendRepository
+from src.presentation.dependencies import get_repository, require_roles
+
+
+router = APIRouter(prefix="/api", tags=["assets"])
+
+
+@router.post(
+    "/articles/{article_id}/assets/generate-specs",
+    response_model=AssetGenerateSpecsResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def post_generate_asset_specs(
+    article_id: UUID,
+    current_user: CurrentUser = Depends(require_roles(EDITOR_OR_ADMIN_ROLES)),
+    repository: BackendRepository = Depends(get_repository),
+) -> AssetGenerateSpecsResponse:
+    try:
+        return generate_asset_specs(
+            repository,
+            article_id=article_id,
+            current_user=current_user,
+        )
+    except LookupError as error:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not Found") from error
+    except PermissionError as error:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(error)) from error
+
+
+@router.get(
+    "/articles/{article_id}/assets",
+    response_model=AssetListResponse,
+)
+def get_article_assets(
+    article_id: UUID,
+    _: CurrentUser = Depends(require_roles(EDITOR_OR_ADMIN_ROLES)),
+    repository: BackendRepository = Depends(get_repository),
+) -> AssetListResponse:
+    try:
+        return list_assets(repository, article_id=article_id)
+    except LookupError as error:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not Found") from error
+
+
+@router.patch(
+    "/articles/{article_id}/assets/{asset_id}",
+    response_model=AssetResponse,
+)
+def patch_article_asset(
+    article_id: UUID,
+    asset_id: UUID,
+    request: AssetUpdateRequest,
+    current_user: CurrentUser = Depends(require_roles(EDITOR_OR_ADMIN_ROLES)),
+    repository: BackendRepository = Depends(get_repository),
+) -> AssetResponse:
+    try:
+        return update_asset(
+            repository,
+            article_id=article_id,
+            asset_id=asset_id,
+            request=request,
+            current_user=current_user,
+        )
+    except LookupError as error:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not Found") from error
+
+
+@router.post(
+    "/articles/{article_id}/assets/{asset_id}/approve",
+    response_model=AssetResponse,
+)
+def post_approve_asset(
+    article_id: UUID,
+    asset_id: UUID,
+    current_user: CurrentUser = Depends(require_roles(EDITOR_OR_ADMIN_ROLES)),
+    repository: BackendRepository = Depends(get_repository),
+) -> AssetResponse:
+    try:
+        return approve_asset(
+            repository,
+            article_id=article_id,
+            asset_id=asset_id,
+            current_user=current_user,
+        )
+    except LookupError as error:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not Found") from error
+
+
+@router.post(
+    "/articles/{article_id}/assets/{asset_id}/upload",
+    response_model=AssetUploadResponse,
+)
+def post_upload_asset_file(
+    article_id: UUID,
+    asset_id: UUID,
+    request: AssetUploadRequest,
+    current_user: CurrentUser = Depends(require_roles(EDITOR_OR_ADMIN_ROLES)),
+    repository: BackendRepository = Depends(get_repository),
+) -> AssetUploadResponse:
+    try:
+        payload = base64.b64decode(request.content_base64.encode("utf-8"), validate=True)
+    except ValueError as error:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Invalid base64 payload",
+        ) from error
+    if not payload:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Uploaded file is empty",
+        )
+    try:
+        return upload_asset_file(
+            repository,
+            article_id=article_id,
+            asset_id=asset_id,
+            filename=request.filename,
+            content_type=request.content_type,
+            content=payload,
+            current_user=current_user,
+            object_storage=open_object_storage_client(),
+        )
+    except LookupError as error:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not Found") from error