Upload 437 files

README.md

@@ -84,6 +84,21 @@ media-api/
 
 The top-level `api/`, `services/`, `operations/`, `workers/`, `core/`, and `models/` packages mirror the canonical `app/` modules as import-compatible entry points for integrations that use the requested layout. Runtime composition uses the single implementation under `app/`, so business logic is not duplicated.
 
+## Current implementation surface
+
+The current source tree exposes the following product areas. Missing docs or backend modules are recorded as omitted from this working tree, not as unimplemented capabilities elsewhere.
+
+- Brand Kit: `/v1/brand`, frontend brand-kit module, Python/TypeScript SDK resources, MCP brand tools, n8n MediaBrandKit node.
+- Projects and collaboration: `/v1/projects`, editor persistence/render jobs, teams/members/invitations, approvals/review comments, notification preferences, Python/TypeScript SDK project resources, MCP collaboration tools, n8n MediaCollaboration node.
+- Publishing and social automation: `/v1/social`, unified publishing, scheduling, retries, reconciliation, analytics sync, Python/TypeScript SDK social resources, MCP social tools, n8n MediaSocial node.
+- Analytics: `/v1/analytics`, overview/timeseries/platform/post analytics, sync runs/cancellation, frontend analytics module, MCP analytics tools, SDK analytics resource. See [`docs/analytics.md`](docs/analytics.md) for implementation and runtime status.
+- AI Studio / Copilot: `/v1/ai`, `/v1/copilot`, durable jobs, deterministic planner, frontend AI/Copilot modules, SDK resources, MCP tools. See [`docs/ai-copilot.md`](docs/ai-copilot.md) for implementation and runtime status.
+- Templates: `/v1/templates`, `/v1/templates/catalog`, template execution, marketplace API, frontend template/marketplace modules, SDK/n8n/MCP coverage. See [`docs/template-marketplace.md`](docs/template-marketplace.md) for implementation and runtime status.
+- Content Studio: editor state, optimistic revisioning, autosave, render jobs, timeline UX, SDK/n8n integration points. See [`docs/content-studio-foundation.md`](docs/content-studio-foundation.md) and [`docs/content-studio-persistence-rendering.md`](docs/content-studio-persistence-rendering.md) for current implementation surface.
+- Brand Kit: `/v1/brand`, frontend brand-kit module, Python/TypeScript SDK resources, MCP brand tools, and an n8n `MediaBrandKit` package entry. See [`docs/brand-kits.md`](docs/brand-kits.md) for implementation and runtime status.
+
+Provider certification, PostgreSQL/RLS runtime, Docker, and Hugging Face runtime verification remain deferred until the planned product phases are complete.
+
 ## Generation foundation
 
 MediaRouter includes a tenant-scoped generation request/job foundation at

api/__init__.py

@@ -1 +1,14 @@
 """Compatibility exports for the canonical :mod:`app.api` package."""
+
+from app.api.api_keys import *  # noqa
+from app.api.audio import *  # noqa
+from app.api.generation import *  # noqa
+from app.api.health import *  # noqa
+from app.api.image import *  # noqa
+from app.api.media import *  # noqa
+from app.api.probe import *  # noqa
+from app.api.social import *  # noqa
+from app.api.templates import *  # noqa
+from app.api.video import *  # noqa
+from app.api.whisper import *  # noqa
+from app.api.ytdlp import *  # noqa

app/api/api_keys.py

@@ -40,7 +40,6 @@ async def current_auth_context(request: Request) -> AuthContextView:
         expires_at=context.expires_at,
         workspace_id=context.workspace_id,
         user_id=context.user_id,
-        membership_role=context.membership_role,
     )
 
 

app/mcp/server.py

@@ -21,6 +21,7 @@ from app.mcp.tools.analytics import register_analytics_tools
 from app.mcp.tools.audio import register_audio_tools
 from app.mcp.tools.brand import register_brand_tools
 from app.mcp.tools.image import register_image_tools
+from app.mcp.tools.collaboration import register_collaboration_tools
 from app.mcp.tools.probe import register_probe_tools
 from app.mcp.tools.social import register_social_tools
 from app.mcp.tools.system import register_system_tools
@@ -62,6 +63,7 @@ def create_mcp_server(container: Container) -> FastMCP[Any]:
     register_template_tools(server, registry)
     register_social_tools(server, registry)
     register_brand_tools(server, registry)
+    register_collaboration_tools(server, registry)
     register_ai_tools(server, registry)
     register_analytics_tools(server, registry)
     register_resources(server, registry)

app/mcp/tools/collaboration.py

@@ -1,25 +1,35 @@
 from __future__ import annotations
 from typing import Any
-from app.mcp.registry import register_tool
+from mcp.server.fastmcp import FastMCP
+
+from app.mcp.registry import MCPRegistry
 from app.projects.services.collaboration_service import CollaborationService
 from app.security.context import auth_context
 
-@register_tool("collaboration.list_teams")
-async def list_teams(ctx: Any) -> list[dict[str, Any]]:
-    """List all teams in the workspace."""
-    auth = auth_context.get()
-    if not auth or not auth.workspace_id:
-        raise Exception("Unauthorized")
-    
-    service: CollaborationService = ctx.container.collaboration
-    return await service.list_teams(auth.workspace_id)
-
-@register_tool("collaboration.create_team")
-async def create_team(ctx: Any, name: str) -> dict[str, Any]:
-    """Create a new team in the workspace."""
-    auth = auth_context.get()
-    if not auth or not auth.workspace_id:
-        raise Exception("Unauthorized")
-    
-    service: CollaborationService = ctx.container.collaboration
-    return (await service.create_team(auth.workspace_id, name)).model_dump()
+
+def register_collaboration_tools(server: FastMCP[Any], registry: MCPRegistry) -> None:
+    @server.tool(description="List teams in the current workspace.")
+    async def list_teams() -> dict[str, Any]:
+        auth = auth_context.get()
+        if not auth or not auth.workspace_id:
+            raise ValueError("Unauthorized")
+
+        async def action() -> list[dict[str, Any]]:
+            service: CollaborationService = registry.container.collaboration
+            return [item.model_dump(mode="json") for item in await service.list_teams(auth.workspace_id)]
+
+        return await registry.run_metadata_tool("collaboration.list_teams", action)
+
+    @server.tool(description="Create a team in the current workspace.")
+    async def create_team(name: str) -> dict[str, Any]:
+        auth = auth_context.get()
+        if not auth or not auth.workspace_id:
+            raise ValueError("Unauthorized")
+
+        async def action() -> dict[str, Any]:
+            service: CollaborationService = registry.container.collaboration
+            return (await service.create_team(auth.workspace_id, name)).model_dump(mode="json")
+
+        return await registry.run_metadata_tool("collaboration.create_team", action)
+
+register_collaboration_tools  # re-export registration symbol for compatibility

app/projects/api.py

@@ -32,7 +32,7 @@ from app.projects.schemas.collaboration import (
     TeamBase, 
     MemberResponse
 )
-from app.projects.schemas.approval import ApprovalRequest, ReviewComment
+from app.projects.schemas.approval import ApprovalRequest, ApprovalRequestCreate, ReviewComment
 from app.security.errors import ForbiddenError
 
 router = APIRouter(prefix="/v1/projects", tags=["projects"])
@@ -82,7 +82,24 @@ async def archive_team(request: Request, team_id: str) -> Response:
 
 @router.get("/workspace/workflows/{workflow_id}/requests", response_model=list[ApprovalRequest])
 async def list_approval_requests(request: Request, workflow_id: str) -> list[ApprovalRequest]:
-    return await request.app.state.container.approval.list_requests(workflow_id)
+    workspace_id, _, _, _ = _identity(request)
+    return await request.app.state.container.approval.list_requests(workspace_id, workflow_id)
+
+
+
+@router.post("/workspace/workflows/{workflow_id}/requests", response_model=ApprovalRequest, status_code=status.HTTP_201_CREATED)
+async def create_approval_request(
+    request: Request,
+    workflow_id: str,
+    payload: ApprovalRequestCreate,
+) -> ApprovalRequest:
+    workspace_id, user_id, _, _ = _identity(request)
+    return await request.app.state.container.approval.create_request(
+        workspace_id=workspace_id,
+        workflow_id=workflow_id,
+        project_id=payload.project_id,
+        user_id=user_id,
+    )
 
 
 @router.post("/workspace/requests/{request_id}/approve", response_model=ApprovalRequest)
@@ -90,16 +107,16 @@ async def approve_request(
     request: Request,
     request_id: str
 ) -> ApprovalRequest:
-    _, user_id, _, _ = _identity(request)
-    return await request.app.state.container.approval.approve_request(request_id, user_id)
+    workspace_id, user_id, _, _ = _identity(request)
+    return await request.app.state.container.approval.approve_request(workspace_id, request_id, user_id)
 
 @router.post("/workspace/requests/{request_id}/reject", response_model=ApprovalRequest)
 async def reject_request(
     request: Request,
     request_id: str
 ) -> ApprovalRequest:
-    _, user_id, _, _ = _identity(request)
-    return await request.app.state.container.approval.reject_request(request_id, user_id)
+    workspace_id, user_id, _, _ = _identity(request)
+    return await request.app.state.container.approval.reject_request(workspace_id, request_id, user_id)
 
 @router.post("/workspace/requests/{request_id}/comments", response_model=ReviewComment)
 async def add_review_comment(
@@ -309,7 +326,10 @@ async def list_project_collaborators(
     request: Request,
     project_id: UUID
 ) -> list[MemberResponse]:
-    return await request.app.state.container.collaboration.list_project_collaborators(str(project_id))
+    workspace_id, user_id, _, _ = _identity(request)
+    return await request.app.state.container.collaboration.list_project_collaborators(
+        workspace_id, str(project_id), user_id=user_id
+    )
 
 @router.post("/{project_id}/collaborators", response_model=MemberResponse)
 async def add_project_collaborator(
@@ -329,7 +349,10 @@ async def remove_project_collaborator(
     project_id: UUID,
     user_id: str
 ) -> Response:
-    await request.app.state.container.collaboration.remove_project_collaborator(str(project_id), user_id)
+    workspace_id, _, _, _ = _identity(request)
+    await request.app.state.container.collaboration.remove_project_collaborator(
+        workspace_id, str(project_id), user_id
+    )
     return Response(status_code=status.HTTP_204_NO_CONTENT)
 
 

app/projects/migrations/0014_approval_workspace_integrity.sql

@@ -0,0 +1,42 @@
+-- Additive approval workspace integrity migration.
+--
+-- This migration brings the existing approval domain in line with the
+-- authoritative ORM models and closes the confirmed cross-workspace
+-- authorization gap by backfilling approval requests with their owning
+-- workflow workspace, enforcing the relationship, and adding an index
+-- used by authorization checks.
+--
+-- Apply after 0013_notification_preferences.sql.
+
+begin;
+
+alter table approval_requests
+  add column if not exists workspace_id text;
+
+do $$
+begin
+  if exists (
+    select 1
+    from approval_requests
+    where workspace_id is null
+  ) then
+    update approval_requests
+    set workspace_id = approval_workflows.workspace_id
+    from approval_workflows
+    where approval_workflows.id = approval_requests.workflow_id
+      and approval_requests.workspace_id is null;
+  end if;
+end $$;
+
+alter table approval_requests
+  alter column workspace_id set not null;
+
+alter table approval_requests
+  drop constraint if exists approval_requests_workflow_id_fkey,
+  add constraint approval_requests_workflow_id_fkey
+    foreign key (workflow_id) references approval_workflows(id) on delete cascade;
+
+create index if not exists ix_approval_requests_workspace
+  on approval_requests(workspace_id);
+
+commit;

app/projects/repositories/approval_repository.py

@@ -30,9 +30,14 @@ class ApprovalRepository:
             await session.refresh(request)
             return request
 
-    async def update_request_status(self, request_id: str, status: str) -> ApprovalRequest:
+    async def update_request_status(self, request_id: str, workspace_id: str, status: str) -> ApprovalRequest:
         async with self.database.session() as session:
-            request = await session.get(ApprovalRequest, request_id)
+            request = await session.scalar(
+                select(ApprovalRequest).where(
+                    ApprovalRequest.id == request_id,
+                    ApprovalRequest.workspace_id == workspace_id,
+                )
+            )
             if not request:
                 raise Exception("Request not found")
             request.status = status
@@ -40,9 +45,14 @@ class ApprovalRepository:
             await session.refresh(request)
             return request
 
-    async def get_request(self, request_id: str) -> ApprovalRequest:
+    async def get_request(self, request_id: str, workspace_id: str) -> ApprovalRequest:
         async with self.database.session() as session:
-            request = await session.get(ApprovalRequest, request_id)
+            request = await session.scalar(
+                select(ApprovalRequest).where(
+                    ApprovalRequest.id == request_id,
+                    ApprovalRequest.workspace_id == workspace_id,
+                )
+            )
             if not request:
                 raise Exception("Request not found")
             return request
@@ -60,10 +70,12 @@ class ApprovalRepository:
             await session.refresh(comment)
             return comment
 
-    async def list_requests(self, workflow_id: str) -> list[ApprovalRequest]:
+    async def list_requests(self, workspace_id: str, workflow_id: str) -> list[ApprovalRequest]:
         async with self.database.session() as session:
             result = await session.scalars(
-                select(ApprovalRequest).where(ApprovalRequest.workflow_id == workflow_id)
+                select(ApprovalRequest).where(
+                    ApprovalRequest.workflow_id == workflow_id,
+                    ApprovalRequest.workspace_id == workspace_id,
+                )
             )
             return list(result.all())
-

app/projects/repositories/collaboration_repository.py

@@ -149,14 +149,23 @@ class CollaborationRepository:
             return list(result.all())
 
     async def list_project_collaborators(self, project_id: str) -> list[ProjectCollaborator]:
-        async with self.database.session() as session:
+        project = await self._project_for_resources(project_id)
+        async with self.database.tenant_session(
+            workspace_id=project.workspace_id, user_id=None
+        ) as session:
             result = await session.scalars(
-                select(ProjectCollaborator).where(ProjectCollaborator.project_id == project_id)
+                select(ProjectCollaborator).where(
+                    ProjectCollaborator.project_id == project_id,
+                    ProjectCollaborator.workspace_id == project.workspace_id,
+                )
             )
             return list(result.all())
 
     async def add_project_collaborator(self, workspace_id: str, project_id: str, user_id: str, role: str) -> ProjectCollaborator:
-        async with self.database.session() as session:
+        await self._project_for_resources(project_id)
+        async with self.database.tenant_session(
+            workspace_id=workspace_id, user_id=user_id
+        ) as session:
             collaborator = ProjectCollaborator(
                 workspace_id=workspace_id,
                 project_id=project_id,
@@ -168,14 +177,28 @@ class CollaborationRepository:
             await session.refresh(collaborator)
             return collaborator
 
-    async def remove_project_collaborator(self, project_id: str, user_id: str) -> None:
-        async with self.database.session() as session:
+    async def remove_project_collaborator(self, workspace_id: str, project_id: str, user_id: str) -> None:
+        await self._project_for_resources(project_id)
+        async with self.database.tenant_session(
+            workspace_id=workspace_id, user_id=user_id
+        ) as session:
             collaborator = await session.scalar(
                 select(ProjectCollaborator).where(
                     ProjectCollaborator.project_id == project_id,
+                    ProjectCollaborator.workspace_id == workspace_id,
                     ProjectCollaborator.user_id == user_id
                 )
             )
             if collaborator:
                 await session.delete(collaborator)
                 await session.commit()
+
+    @staticmethod
+    async def _project_for_resources(project_id: str) -> Project:
+        async with CollaborationRepository(None).database.session() as session:
+            project = await session.scalar(
+                select(Project).where(Project.id == project_id)
+            )
+            if project is None:
+                raise Exception("Project not found")
+            return project

app/projects/schemas/approval.py

@@ -9,6 +9,11 @@ class ApprovalRequest(BaseModel):
     created_by: str
     created_at: str
 
+
+class ApprovalRequestCreate(BaseModel):
+    project_id: str
+
+
 class ReviewComment(BaseModel):
     id: str
     request_id: str

app/projects/schemas/collaboration.py

@@ -20,7 +20,6 @@ class InvitationResponse(InvitationCreate):
     status: str
     expires_at: str
     created_at: str
-    token: str | None = None
 
 class TeamBase(BaseModel):
     name: str

app/projects/services/approval_service.py

@@ -33,5 +33,5 @@ class ApprovalService:
     async def add_comment(self, request_id: str, user_id: str, workspace_id: str, content: str) -> ReviewComment:
         return await self.repository.add_review_comment(request_id, user_id, workspace_id, content)
 
-    async def list_requests(self, workflow_id: str) -> list[ApprovalRequest]:
-        return await self.repository.list_requests(workflow_id)
+    async def list_requests(self, workspace_id: str, workflow_id: str) -> list[ApprovalRequest]:
+        return await self.repository.list_requests(workspace_id, workflow_id)

app/projects/services/collaboration_service.py

@@ -50,7 +50,6 @@ class CollaborationService:
             expires_at=invitation.expires_at.isoformat(),
             created_at=invitation.created_at.isoformat()
         )
-        setattr(response, "token", token) 
         return response
 
     async def list_members(self, workspace_id: str) -> list[MemberResponse]:
@@ -107,14 +106,17 @@ class CollaborationService:
     async def get_membership(self, workspace_id: str, user_id: str) -> WorkspaceMembership | None:
         return await self.repository.get_membership(workspace_id, user_id)
 
-    async def list_project_collaborators(self, project_id: str) -> list[MemberResponse]:
-        collaborators = await self.repository.list_project_collaborators(project_id)
+    async def list_project_collaborators(self, workspace_id: str, project_id: str, *, user_id: str) -> list[MemberResponse]:
+        collaborators = await self.repository.list_project_collaborators(workspace_id, project_id)
         return [MemberResponse(id=c.id, workspace_id=c.workspace_id, user_id=c.user_id, role=c.role, created_at=c.created_at.isoformat()) for c in collaborators]
 
     async def add_project_collaborator(self, workspace_id: str, project_id: str, user_id: str, role: str) -> MemberResponse:
         collaborator = await self.repository.add_project_collaborator(workspace_id, project_id, user_id, role)
         return MemberResponse(id=collaborator.id, workspace_id=collaborator.workspace_id, user_id=collaborator.user_id, role=collaborator.role, created_at=collaborator.created_at.isoformat())
 
+    async def remove_project_collaborator(self, workspace_id: str, project_id: str, user_id: str) -> None:
+        await self.repository.remove_project_collaborator(workspace_id, project_id, user_id)
+
     async def record_activity(self, workspace_id: str, user_id: str, action: str, entity_id: str, entity_type: str, metadata: dict[str, Any]) -> None:
         await self.repository.record_activity(workspace_id, user_id, action, entity_id, entity_type, metadata)
 

app/security/schemas.py

@@ -133,7 +133,6 @@ class AuthContextView(BaseModel):
     expires_at: datetime | None
     workspace_id: str | None = None
     user_id: str | None = None
-    membership_role: str | None = None
 
 
 class AuditLogView(BaseModel):

core/__init__.py

@@ -1 +1,6 @@
 """Compatibility exports for the canonical :mod:`app.core` package."""
+
+from app.core.config import *  # noqa
+from app.core.exceptions import *  # noqa
+from app.core.logger import *  # noqa
+from app.core.response import *  # noqa

models/__init__.py

@@ -1 +1,4 @@
 """Compatibility exports for the canonical :mod:`app.models` package."""
+
+from app.models.media import *  # noqa
+from app.models.requests import *  # noqa

operations/__init__.py

@@ -1 +1,15 @@
 """Compatibility exports for the canonical :mod:`app.operations` package."""
+
+from app.operations.common import *  # noqa
+from app.operations.compress import *  # noqa
+from app.operations.concat import *  # noqa
+from app.operations.convert import *  # noqa
+from app.operations.crop import *  # noqa
+from app.operations.extract_audio import *  # noqa
+from app.operations.merge import *  # noqa
+from app.operations.resize import *  # noqa
+from app.operations.rotate import *  # noqa
+from app.operations.subtitles import *  # noqa
+from app.operations.thumbnails import *  # noqa
+from app.operations.trim import *  # noqa
+from app.operations.watermark import *  # noqa

services/__init__.py

@@ -0,0 +1,11 @@
+"""Compatibility exports for the canonical :mod:`app.services` package."""
+
+from app.services.cleanup import *  # noqa
+from app.services.downloader import *  # noqa
+from app.services.ffmpeg_service import *  # noqa
+from app.services.ffprobe_service import *  # noqa
+from app.services.input_resolver import *  # noqa
+from app.services.media_service import *  # noqa
+from app.services.validator import *  # noqa
+from app.services.whisper_service import *  # noqa
+from app.services.ytdlp_service import *  # noqa

services/cleanup.py

@@ -0,0 +1 @@
+from app.services.cleanup import *  # noqa

services/downloader.py

@@ -0,0 +1 @@
+from app.services.downloader import *  # noqa

services/ffmpeg_service.py

@@ -0,0 +1 @@
+from app.services.ffmpeg_service import *  # noqa

services/ffprobe_service.py

@@ -0,0 +1 @@
+from app.services.ffprobe_service import *  # noqa

services/input_resolver.py

@@ -0,0 +1 @@
+from app.services.input_resolver import *  # noqa

services/media_service.py

@@ -0,0 +1 @@
+from app.services.media_service import *  # noqa

services/validator.py

@@ -0,0 +1 @@
+from app.services.validator import *  # noqa

services/whisper_service.py

@@ -0,0 +1 @@
+from app.services.whisper_service import *  # noqa

services/ytdlp_service.py

@@ -0,0 +1 @@
+from app.services.ytdlp_service import *  # noqa

tests/conftest.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from app.core.config import Settings
+
+
+@pytest.fixture
+def settings(tmp_path: Path) -> Settings:
+    return Settings(
+        _env_file=None,
+        temp_dir=tmp_path / "temp",
+        output_dir=tmp_path / "outputs",
+        max_upload_size=10 * 1024 * 1024,
+        cleanup_minutes=1,
+        cleanup_interval_seconds=3600,
+        whisper_model="tiny",
+        max_workers=1,
+        allow_private_urls=True,
+        auth_enabled=False,
+        database_url=f"sqlite+aiosqlite:///{tmp_path / 'security.db'}",
+    )

tests/test_ai_copilot.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+from pydantic import ValidationError
+
+from app.copilot.actions import CopilotActionRegistry
+from app.copilot.errors import CopilotInvalidRequestError
+from app.copilot.planner import CopilotPlanner
+from app.copilot.schemas import (
+    CopilotContext,
+    CopilotEditorSummary,
+    CopilotPlan,
+)
+
+
+def context(*, capabilities: list[str], asset: bool = False, clip: bool = False):
+    project_id = uuid4()
+    return CopilotContext(
+        workspace_id=str(uuid4()),
+        project_id=project_id,
+        selected_asset_ids=[uuid4()] if asset else [],
+        selected_clip_ids=["clip-1"] if clip else [],
+        editor_summary=CopilotEditorSummary(
+            revision=4, duration_ms=30_000, track_count=1, clip_count=1
+        ),
+        available_capabilities=capabilities,
+    )
+
+
+def test_planner_fails_closed_for_unavailable_transcription() -> None:
+    plan = CopilotPlanner().plan(
+        "Turn this podcast into a TikTok",
+        context(capabilities=["editor.render"], asset=True),
+    )
+    assert not plan.executable
+    assert plan.unsupported_capabilities == ["ai.transcribe"]
+    assert plan.actions == []
+
+
+def test_planner_requires_confirmation_for_render_and_generation() -> None:
+    render = CopilotPlanner().plan("Render this project", context(capabilities=["editor.render"]))
+    assert render.executable and render.requires_confirmation
+    assert render.actions[0].type == "editor.render"
+    image = CopilotPlanner().plan(
+        "Generate an image of a lighthouse",
+        context(capabilities=["ai.generate_image"]),
+    )
+    assert image.executable and image.requires_confirmation
+    assert image.actions[0].type == "ai.generate_image"
+
+
+def test_action_plan_rejects_unknown_model_generated_structures() -> None:
+    with pytest.raises(ValidationError):
+        CopilotPlan.model_validate(
+            {
+                "intent": "unsafe",
+                "explanation": "unsafe",
+                "actions": [
+                    {
+                        "id": "a",
+                        "type": "shell.execute",
+                        "arguments": {"command": "rm -rf /"},
+                        "reason": "unsafe",
+                        "requires_confirmation": False,
+                        "destructive": False,
+                        "external_side_effect": False,
+                        "required_permission": "admin",
+                        "required_capability": "shell",
+                    }
+                ],
+                "missing_information": [],
+                "unsupported_capabilities": [],
+                "executable": True,
+                "requires_confirmation": False,
+            }
+        )
+
+
+def test_action_registry_rejects_policy_metadata_tampering() -> None:
+    registry = CopilotActionRegistry(
+        projects=None,  # type: ignore[arg-type]
+        assets=None,  # type: ignore[arg-type]
+        editor=None,  # type: ignore[arg-type]
+        renders=None,  # type: ignore[arg-type]
+        ai=None,  # type: ignore[arg-type]
+        templates=None,  # type: ignore[arg-type]
+    )
+    plan = CopilotPlanner().plan(
+        "Generate an image of a lighthouse",
+        context(capabilities=["ai.generate_image"]),
+    )
+    tampered = plan.actions[0].model_copy(update={"requires_confirmation": False})
+    with pytest.raises(CopilotInvalidRequestError):
+        registry.validate(tampered)
+
+
+def test_copilot_migration_is_additive_and_tenant_isolated() -> None:
+    migration = (
+        (Path(__file__).resolve().parents[1] / "app/projects/migrations/0005_ai_copilot.sql")
+        .read_text(encoding="utf-8")
+        .lower()
+    )
+    for expected in (
+        "create table if not exists copilot_runs",
+        "unique (workspace_id, idempotency_key)",
+        "enable row level security",
+        "force row level security",
+        "create policy copilot_runs_select",
+        "create policy copilot_runs_insert",
+        "create policy copilot_runs_update",
+        "copilot run identity fields are immutable",
+    ):
+        assert expected in migration
+    assert "drop table" not in migration

tests/test_analytics_phase10_static.py

@@ -0,0 +1,42 @@
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_analytics_migration_is_additive_and_forces_rls() -> None:
+    text = (ROOT / "app/social/migrations/0010_analytics_insights.sql").read_text()
+    normalized = text.lower()
+    assert "drop table" not in normalized
+    for table in (
+        "analytics_sync_runs",
+        "analytics_metric_snapshots",
+        "analytics_post_metrics",
+        "analytics_platform_metrics",
+    ):
+        assert f"create table if not exists {table}" in normalized
+    assert normalized.count("force row level security") >= 1
+    assert "current_setting(''app.workspace_id''" in normalized
+
+
+def test_analytics_routes_and_transports_are_narrow() -> None:
+    api = (ROOT / "app/analytics/api.py").read_text()
+    mcp = (ROOT / "app/mcp/tools/analytics.py").read_text()
+    sdk = (ROOT / "sdk/typescript/src/resources/analytics.ts").read_text()
+    for route in (
+        '"/overview"',
+        '"/timeseries"',
+        '"/platforms"',
+        '"/posts"',
+        '"/sync"',
+    ):
+        assert route in api
+    assert "execute analytics query" not in mcp.lower()
+    assert "class AnalyticsResource" in sdk
+
+
+def test_analytics_never_fabricates_provider_metrics() -> None:
+    service = (ROOT / "app/analytics/service.py").read_text()
+    provider = (ROOT / "app/social/providers/base.py").read_text()
+    assert "self.social_analytics.post" in service
+    assert "get_metrics" in provider
+    assert "random.randint" not in service

tests/test_api_contract_regression.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import ast
+import re
+from pathlib import Path
+import unittest
+
+
+ROUTES = {
+    "brand_api": Path("app/brand/api.py"),
+    "projects_api": Path("app/projects/api.py"),
+}
+
+FRONTEND_CALLS = {
+    "brand_api": Path("frontend/features/brand-kits/api/index.ts"),
+    "collaboration_api": Path("frontend/features/workspace/collaboration/api/collaboration.ts"),
+}
+
+BRAND_ROUTES = {
+    "router.post('', response_model=BrandKitResponse, status_code=status.HTTP_201_CREATED)": "/v1/brand POST",
+    "router.get('', response_model=list[BrandKitResponse])": "/v1/brand GET",
+    "router.patch('/{brand_kit_id}', response_model=BrandKitResponse)": "/v1/brand PATCH",
+    "router.delete('/{brand_kit_id}', status_code=status.HTTP_204_NO_CONTENT)": "/v1/brand DELETE",
+}
+
+EXPECTED_BRAND_FRONTEND_CALLS = [
+    "await apiClient.get('/v1/brand');",
+    "await apiClient.post('/v1/brand', payload);",
+]
+
+EXPECTED_COLLABORATION_FRONTEND_CALLS = [
+    "await apiClient.get('/v1/projects/workspace/teams');",
+    "await apiClient.post('/v1/projects/workspace/teams', payload);",
+    "await apiClient.post('/v1/projects/workspace/invitations', payload);",
+    "await apiClient.get('/v1/projects/workspace/members');",
+    "await apiClient.delete(`/v1/projects/workspace/members/${userId}`);",
+    "await apiClient.patch(`/v1/projects/workspace/members/${userId}/role?new_role=${newRole}`);",
+    "await apiClient.get(`/v1/projects/workspace/workflows/${workflowId}/requests`);",
+    "await apiClient.post(`/v1/projects/workspace/workflows/${workflowId}/requests`, { project_id: projectId });",
+    "await apiClient.post(`/v1/projects/workspace/requests/${requestId}/approve`);",
+    "await apiClient.post(`/v1/projects/workspace/requests/${requestId}/reject`);",
+    "await apiClient.post(`/v1/projects/workspace/requests/${requestId}/comments?content=${encodeURIComponent(content)}`);",
+    "await apiClient.get(`/v1/projects/${encodeURIComponent(projectId)}/collaborators`);",
+    "await apiClient.post(`/v1/projects/${encodeURIComponent(projectId)}/collaborators?user_id=${encodeURIComponent(userId)}&role=${encodeURIComponent(role)}`);",
+    "await apiClient.delete(`/v1/projects/${encodeURIComponent(projectId)}/collaborators/${encodeURIComponent(userId)}`);",
+]
+
+
+def _route_decorators(path: Path) -> list[str]:
+    tree = ast.parse(path.read_text())
+    calls = []
+    for node in tree.body:
+        if not isinstance(node, ast.AsyncFunctionDef):
+            continue
+        for decorator in node.decorator_list:
+            if isinstance(decorator, ast.Call) and isinstance(decorator.func, ast.Attribute) and decorator.func.attr in {"get", "post", "patch", "delete"}:
+                calls.append(ast.unparse(decorator))
+    return calls
+
+
+def _frontend_calls(path: Path) -> list[str]:
+    return [re.sub(r"^\s*const\s+\{[^}]*\}\s+=\s+", "", line.strip()) for line in path.read_text().splitlines() if "apiClient." in line]
+
+
+def test_brand_kit_routes_match_expected_contract() -> None:
+    assert _route_decorators(ROUTES["brand_api"]) == list(BRAND_ROUTES.keys())
+
+
+def test_approval_request_create_route_is_exposed() -> None:
+    decorators = _route_decorators(ROUTES["projects_api"])
+    assert any(
+        decorator == "router.post('/workspace/workflows/{workflow_id}/requests', response_model=ApprovalRequest, status_code=status.HTTP_201_CREATED)"
+        for decorator in decorators
+    )
+
+
+def test_brand_kit_frontend_uses_expected_backend_routes() -> None:
+    assert _frontend_calls(FRONTEND_CALLS["brand_api"]) == EXPECTED_BRAND_FRONTEND_CALLS
+
+
+def test_collaboration_frontend_uses_expected_backend_routes() -> None:
+    assert _frontend_calls(FRONTEND_CALLS["collaboration_api"]) == EXPECTED_COLLABORATION_FRONTEND_CALLS
+
+
+class ApiContractRegressionTests(unittest.TestCase):
+    def test_brand_kit_routes_match_expected_contract(self) -> None:
+        test_brand_kit_routes_match_expected_contract()
+
+    def test_brand_kit_frontend_uses_expected_backend_routes(self) -> None:
+        test_brand_kit_frontend_uses_expected_backend_routes()
+
+    def test_collaboration_frontend_uses_expected_backend_routes(self) -> None:
+        test_collaboration_frontend_uses_expected_backend_routes()
+
+    def test_approval_request_create_route_is_exposed(self) -> None:
+        test_approval_request_create_route_is_exposed()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_authentication.py

@@ -0,0 +1,307 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from app.container import build_container
+from app.core.config import Settings
+from app.mcp.registry import MCPRegistry
+from app.security.context import auth_context
+from app.security.errors import APIKeyConflictError, ForbiddenError, RateLimitError, UnauthorizedError
+from app.security.models import APIKey, AuditLog
+from app.security.schemas import APIKeyCreate
+from app.security.service import APIKeyService
+from main import create_app
+
+
+def security_settings(tmp_path: Path, **overrides: object) -> Settings:
+    values: dict[str, object] = {
+        "_env_file": None,
+        "temp_dir": tmp_path / "temp",
+        "output_dir": tmp_path / "outputs",
+        "database_url": f"sqlite+aiosqlite:///{tmp_path / 'security.db'}",
+        "auth_enabled": True,
+        "auth_last_used_update_seconds": 0,
+        "cleanup_interval_seconds": 3600,
+        "whisper_model": "tiny",
+        "max_workers": 1,
+    }
+    values.update(overrides)
+    return Settings(**values)
+
+
+@pytest.fixture
+async def security_container(tmp_path: Path):
+    container = build_container(security_settings(tmp_path))
+    await container.security_database.initialize()
+    try:
+        yield container
+    finally:
+        await container.security_database.close()
+
+
+async def create_key(container, **overrides: object) -> tuple[APIKey, str]:
+    values: dict[str, object] = {
+        "name": "Automation",
+        "environment": "test",
+        "role": None,
+        "scopes": ["templates:read"],
+    }
+    values.update(overrides)
+    return await container.api_keys.create(APIKeyCreate(**values), created_by="tests")
+
+
+async def test_key_generation_has_256_bits_and_database_never_stores_secret(
+    security_container,
+) -> None:
+    record, secret = await create_key(security_container)
+
+    environment, encoded_secret = secret.split("_", 2)[1:]
+    raw_secret = base64.urlsafe_b64decode(encoded_secret + "=")
+    assert environment == "test"
+    assert len(raw_secret) == 32
+    assert record.key_prefix == f"mp_test_{encoded_secret[:8]}"
+    assert record.key_hash == hashlib.sha256(secret.encode()).hexdigest()
+
+    async with security_container.security_database.session() as session:
+        stored = await session.get(APIKey, record.id)
+        assert stored is not None
+        assert secret not in vars(stored).values()
+        assert not hasattr(stored, "api_key")
+
+
+async def test_authentication_rejects_invalid_expired_disabled_and_revoked_keys(
+    security_container,
+) -> None:
+    active, active_secret = await create_key(security_container)
+    assert (await security_container.api_keys.authenticate(active_secret)).api_key_id == active.id
+
+    replacement = "A" if active_secret[-1] != "A" else "B"
+    with pytest.raises(UnauthorizedError):
+        await security_container.api_keys.authenticate(active_secret[:-1] + replacement)
+
+    _, expired_secret = await create_key(
+        security_container,
+        name="Expired",
+        expires_at=datetime.now(timezone.utc) - timedelta(seconds=1),
+    )
+    with pytest.raises(UnauthorizedError):
+        await security_container.api_keys.authenticate(expired_secret)
+
+    await security_container.api_keys.set_status(active.id, "disabled")
+    with pytest.raises(UnauthorizedError):
+        await security_container.api_keys.authenticate(active_secret)
+    await security_container.api_keys.set_status(active.id, "active")
+    assert (await security_container.api_keys.authenticate(active_secret)).api_key_id == active.id
+
+    await security_container.api_keys.set_status(active.id, "revoked")
+    with pytest.raises(UnauthorizedError):
+        await security_container.api_keys.authenticate(active_secret)
+    with pytest.raises(APIKeyConflictError):
+        await security_container.api_keys.set_status(active.id, "disabled")
+    with pytest.raises(APIKeyConflictError):
+        await security_container.api_keys.set_status(active.id, "active")
+
+
+async def test_scope_enforcement_and_rotation_grace_period(security_container) -> None:
+    old, old_secret = await create_key(security_container)
+    context = await security_container.api_keys.authenticate(old_secret)
+    security_container.api_keys.authorize(context, "templates:read")
+    with pytest.raises(ForbiddenError):
+        security_container.api_keys.authorize(context, "operations:execute")
+
+    replacement, replacement_secret = await security_container.api_keys.rotate(
+        old.id, 60, created_by="tests"
+    )
+    assert replacement.rotated_from_id == old.id
+    assert (await security_container.api_keys.authenticate(old_secret)).api_key_id == old.id
+    assert (
+        await security_container.api_keys.authenticate(replacement_secret)
+    ).api_key_id == replacement.id
+    with pytest.raises(APIKeyConflictError):
+        await security_container.api_keys.set_status(old.id, "disabled")
+
+    async with security_container.security_database.session() as session:
+        rotating = await session.get(APIKey, old.id)
+        assert rotating is not None
+        rotating.grace_expires_at = datetime.now(timezone.utc) - timedelta(seconds=1)
+        await session.commit()
+    with pytest.raises(UnauthorizedError):
+        await security_container.api_keys.authenticate(old_secret)
+    assert (await security_container.api_keys.get(old.id)).status == "revoked"
+
+
+async def test_per_key_request_and_concurrent_job_limits(security_container) -> None:
+    _, request_secret = await create_key(
+        security_container, name="RPM", requests_per_minute=1
+    )
+    request_context = await security_container.api_keys.authenticate(request_secret)
+    lease = await security_container.rate_limiter.acquire(
+        request_context, is_job=False, is_upload=False, uploaded_bytes=0
+    )
+    await lease.release()
+    with pytest.raises(RateLimitError) as rate_error:
+        await security_container.rate_limiter.acquire(
+            request_context, is_job=False, is_upload=False, uploaded_bytes=0
+        )
+    assert rate_error.value.retry_after >= 1
+
+    _, job_secret = await create_key(
+        security_container, name="Concurrency", concurrent_jobs=1
+    )
+    job_context = await security_container.api_keys.authenticate(job_secret)
+    running = await security_container.rate_limiter.acquire(
+        job_context, is_job=True, is_upload=False, uploaded_bytes=0
+    )
+    with pytest.raises(RateLimitError):
+        await security_container.rate_limiter.acquire(
+            job_context, is_job=True, is_upload=False, uploaded_bytes=0
+        )
+    await running.release()
+    next_job = await security_container.rate_limiter.acquire(
+        job_context, is_job=True, is_upload=False, uploaded_bytes=0
+    )
+    await next_job.release()
+
+
+async def test_stdio_mcp_uses_shared_context_scopes_rate_limits_and_audit(
+    security_container,
+) -> None:
+    _, secret = await create_key(
+        security_container, name="MCP Reader", scopes=["mcp:read"]
+    )
+    context = await security_container.api_keys.authenticate(secret)
+    registry = MCPRegistry(security_container)
+    unauthorized = await registry.run_metadata_tool("system_info", registry.system_info_data)
+    token = auth_context.set(context)
+    try:
+        resource = await registry.safe_resource("version", registry.version_data)
+        forbidden = await registry.run_metadata_tool("system_info", registry.system_info_data)
+    finally:
+        auth_context.reset(token)
+
+    assert unauthorized["success"] is False
+    assert unauthorized["error"]["code"] == "UNAUTHORIZED"
+    assert resource["success"] is True
+    assert forbidden["success"] is False
+    assert forbidden["error"]["code"] == "FORBIDDEN"
+    async with security_container.security_database.session() as session:
+        logs = list((await session.scalars(select(AuditLog))).all())
+    assert {log.endpoint for log in logs} >= {
+        "mcp://tools/resource.version",
+        "mcp://tools/system_info",
+    }
+
+
+def test_http_middleware_public_and_authentication_contracts(tmp_path: Path) -> None:
+    material = APIKeyService.generate_material("test")
+    settings = security_settings(
+        tmp_path,
+        auth_bootstrap_key_hash=material.key_hash,
+        auth_bootstrap_key_prefix=material.key_prefix,
+        auth_bootstrap_environment="test",
+        auth_default_requests_per_minute=1000,
+    )
+    application = create_app(settings)
+    authorization = {"Authorization": f"Bearer {material.api_key}"}
+
+    with TestClient(application) as client:
+        for path in ("/", "/health", "/version", "/docs", "/openapi.json", "/redoc"):
+            assert client.get(path).status_code == 200
+
+        missing = client.get("/v1/auth/context")
+        malformed = client.get(
+            "/v1/auth/context", headers={"Authorization": "Basic not-a-mediarouter-key"}
+        )
+        invalid = client.get(
+            "/v1/auth/context", headers={"Authorization": "Bearer mp_test_invalid"}
+        )
+        for response in (missing, malformed, invalid):
+            assert response.status_code == 401
+            assert response.json() == {
+                "error": "Unauthorized",
+                "message": "Invalid or expired API key.",
+            }
+            assert response.headers["www-authenticate"] == "Bearer"
+
+        mcp_missing = client.post("/mcp/", json={"jsonrpc": "2.0", "id": 1})
+        assert mcp_missing.status_code == 401
+
+        identity = client.get("/v1/auth/context", headers=authorization)
+        assert identity.status_code == 200
+        assert identity.json()["key_prefix"] == material.key_prefix
+        assert "admin" in identity.json()["scopes"]
+
+        created = client.post(
+            "/v1/api-keys",
+            headers=authorization,
+            json={
+                "name": "Template Reader",
+                "environment": "test",
+                "role": None,
+                "scopes": ["templates:read"],
+            },
+        )
+        assert created.status_code == 201
+        limited_authorization = {
+            "Authorization": f"Bearer {created.json()['api_key']}"
+        }
+        assert client.get("/v1/auth/context", headers=limited_authorization).status_code == 200
+        forbidden = client.get("/v1/health", headers=limited_authorization)
+        assert forbidden.status_code == 403
+        assert forbidden.json() == {
+            "error": "Forbidden",
+            "message": "Missing required scope.",
+        }
+
+        mcp_forbidden = client.post(
+            "/mcp/",
+            headers=limited_authorization,
+            json={
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "tools/call",
+                "params": {"name": "health", "arguments": {}},
+            },
+        )
+        assert mcp_forbidden.status_code == 403
+
+        audit_logs = client.get("/v1/audit-logs", headers=authorization)
+        assert audit_logs.status_code == 200
+        entries = audit_logs.json()
+        assert any(
+            entry["endpoint"] == "/v1/auth/context"
+            and entry["api_key_id"] == identity.json()["id"]
+            and entry["response_code"] == 200
+            for entry in entries
+        )
+
+
+def test_http_rate_limit_returns_retry_after(tmp_path: Path) -> None:
+    material = APIKeyService.generate_material("test")
+    application = create_app(
+        security_settings(
+            tmp_path,
+            auth_bootstrap_key_hash=material.key_hash,
+            auth_bootstrap_key_prefix=material.key_prefix,
+            auth_bootstrap_environment="test",
+            auth_default_requests_per_minute=1,
+        )
+    )
+    headers = {"Authorization": f"Bearer {material.api_key}"}
+    with TestClient(application) as client:
+        assert client.get("/v1/auth/context", headers=headers).status_code == 200
+        limited = client.get("/v1/auth/context", headers=headers)
+
+    assert limited.status_code == 429
+    assert limited.json() == {
+        "error": "Rate limit exceeded",
+        "message": "Retry later.",
+    }
+    assert int(limited.headers["retry-after"]) >= 1

tests/test_brand_kits.py

@@ -0,0 +1,42 @@
+import pytest
+from unittest.mock import AsyncMock, MagicMock
+from app.brand.services.brand_service import BrandKitService
+from app.brand.services.validation_service import BrandKitValidationService
+from app.brand.models.brand import BrandKitVersion
+
+@pytest.fixture
+def validation_service():
+    return BrandKitValidationService()
+
+def test_brand_kit_validation_missing_logo(validation_service):
+    version = BrandKitVersion(version_number=1, created_by="user1")
+    result = validation_service.validate(version)
+    assert not result['valid']
+    assert any(issue['field'] == 'logo_asset_id' for issue in result['issues'])
+
+def test_brand_kit_validation_valid(validation_service):
+    version = BrandKitVersion(version_number=1, created_by="user1", logo_asset_id="asset123")
+    result = validation_service.validate(version)
+    assert result['valid']
+
+@pytest.mark.asyncio
+async def test_brand_kit_service_create():
+    mock_repo = AsyncMock()
+    mock_assets = AsyncMock()
+    mock_audit = AsyncMock()
+    
+    service = BrandKitService(mock_repo, mock_assets, mock_audit)
+    
+    workspace_id = "ws1"
+    name = "Test Kit"
+    data = {"logo_asset_id": "asset123"}
+    user_id = "user1"
+    
+    mock_assets.get_asset.return_value = {"id": "asset123"}
+    mock_repo.create.return_value = (MagicMock(id="kit1"), MagicMock(id="ver1"))
+    
+    await service.create_brand_kit(workspace_id, name, data, user_id=user_id)
+    
+    mock_assets.get_asset.assert_called_once()
+    mock_repo.create.assert_called_once()
+    mock_audit.log_event.assert_called_once()

tests/test_cleanup_worker.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import asyncio
+import os
+import time
+from uuid import uuid4
+
+from app.services.cleanup import CleanupService
+from app.workers.cleanup_worker import CleanupWorker
+
+
+async def test_cleanup_removes_expired_workspace(settings) -> None:
+    service = CleanupService(settings)
+    request_id = str(uuid4())
+    workspace = await service.create_workspace(request_id)
+    await service.complete(request_id)
+    old = time.time() - 120
+    os.utime(workspace.root, (old, old))
+    removed = await service.cleanup_expired()
+    assert removed == 1
+    assert not workspace.root.exists()
+
+
+async def test_cleanup_keeps_active_workspace(settings) -> None:
+    service = CleanupService(settings)
+    workspace = await service.create_workspace(str(uuid4()))
+    old = time.time() - 120
+    os.utime(workspace.root, (old, old))
+    assert await service.cleanup_expired() == 0
+    assert workspace.root.exists()
+
+
+async def test_cleanup_worker_runs_and_stops() -> None:
+    class FakeCleanup:
+        def __init__(self) -> None:
+            self.called = asyncio.Event()
+
+        async def cleanup_expired(self) -> int:
+            self.called.set()
+            return 0
+
+    service = FakeCleanup()
+    worker = CleanupWorker(service, interval_seconds=60)  # type: ignore[arg-type]
+    await worker.start()
+    await asyncio.wait_for(service.called.wait(), timeout=1)
+    await worker.stop()

tests/test_collaboration.py

@@ -0,0 +1,33 @@
+import pytest
+from app.projects.repositories.collaboration_repository import CollaborationRepository
+from app.projects.services.collaboration_service import CollaborationService
+from app.projects.errors import CollaborationUnauthorizedError
+
+@pytest.mark.asyncio
+async def test_collaboration_logic_admin_removal_constraint(db_session):
+    # Setup test workspace and admin users
+    repo = CollaborationRepository(db_session)
+    service = CollaborationService(repo)
+    
+    workspace_id = "test_workspace"
+    admin_user_id = "admin_user"
+    target_user_id = "member_user"
+    
+    # 1. Mock memberships: 2 admins
+    # Use actual DB insert here if needed for true integration test
+    # ... setup DB state ...
+
+    # 2. Test prevention of last admin removal
+    with pytest.raises(CollaborationUnauthorizedError, match="Cannot remove the last administrator."):
+        await service.remove_member(workspace_id, admin_user_id, target_user_id)
+
+@pytest.mark.asyncio
+async def test_collaboration_logic_self_elevation_prevention(db_session):
+    repo = CollaborationRepository(db_session)
+    service = CollaborationService(repo)
+    
+    workspace_id = "test_workspace"
+    actor_user_id = "user_1"
+    
+    with pytest.raises(CollaborationUnauthorizedError, match="Cannot elevate your own privileges."):
+        await service.update_member_role(workspace_id, actor_user_id, actor_user_id, "admin")

tests/test_collaboration_full.py

@@ -0,0 +1,40 @@
+import pytest
+from app.projects.repositories.collaboration_repository import CollaborationRepository
+from app.projects.services.collaboration_service import CollaborationService
+
+@pytest.mark.asyncio
+async def test_collaboration_team_lifecycle(db_session):
+    repo = CollaborationRepository(db_session)
+    service = CollaborationService(repo)
+    
+    workspace_id = "test_workspace"
+    
+    # 1. Create Team
+    team = await service.create_team(workspace_id, "Engineering")
+    assert team.name == "Engineering"
+    
+    # 2. List Teams
+    teams = await service.list_teams(workspace_id)
+    assert len(teams) >= 1
+    
+    # 3. Update Team
+    updated = await service.update_team(workspace_id, team.id, "Product")
+    assert updated.name == "Product"
+    
+    # 4. Archive
+    await service.archive_team(workspace_id, team.id)
+    teams = await service.list_teams(workspace_id)
+    assert not any(t.id == team.id for t in teams)
+
+@pytest.mark.asyncio
+async def test_collaboration_invitation_lifecycle(db_session):
+    repo = CollaborationRepository(db_session)
+    service = CollaborationService(repo)
+    
+    workspace_id = "test_workspace"
+    email = "test@example.com"
+    
+    # Test invitation
+    invitation = await service.invite_member(workspace_id, email, "member")
+    assert invitation.email == email
+    assert hasattr(invitation, "token") # Check if token is returned

tests/test_content_studio_phase2.py

@@ -0,0 +1,295 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+from sqlalchemy import select
+
+from app.container import build_container
+from app.core.config import Settings
+from app.projects.editor_schemas import EditorDocument, EditorSaveRequest, ProjectRenderCreate
+from app.projects.errors import (
+    ProjectEditorConflictError,
+    ProjectNotFoundError,
+    ProjectRenderLimitError,
+)
+from app.projects.schemas import ProjectCreate
+from app.projects.services.render_compiler import compile_render
+from app.security.models import AuditEvent
+from app.security.schemas import APIKeyCreate
+
+
+def settings(tmp_path: Path) -> Settings:
+    return Settings(
+        _env_file=None,
+        auth_enabled=True,
+        database_url=f"sqlite+aiosqlite:///{tmp_path / 'security.db'}",
+        social_database_url=f"sqlite+aiosqlite:///{tmp_path / 'social.db'}",
+        social_auto_migrate=True,
+        social_worker_enabled=False,
+        generation_worker_enabled=False,
+        render_worker_enabled=False,
+        social_oauth_encryption_key="test-only-encryption-material",
+        temp_dir=tmp_path / "temp",
+        output_dir=tmp_path / "outputs",
+        whisper_model="tiny",
+    )
+
+
+def document(project_id: str, asset_id: str) -> EditorDocument:
+    return EditorDocument.model_validate(
+        {
+            "schemaVersion": 1,
+            "projectId": project_id,
+            "timeline": {
+                "timeUnit": "milliseconds",
+                "tracks": [
+                    {
+                        "id": "video-1",
+                        "type": "video",
+                        "name": "Video 1",
+                        "order": 0,
+                        "muted": False,
+                        "locked": False,
+                        "visible": True,
+                        "clips": [
+                            {
+                                "id": "clip-1",
+                                "kind": "media",
+                                "trackId": "video-1",
+                                "assetId": asset_id,
+                                "label": "source.mp4",
+                                "startMs": 0,
+                                "durationMs": 1000,
+                                "sourceStartMs": 0,
+                                "sourceDurationMs": 1000,
+                                "mediaType": "video",
+                                "transform": {
+                                    "x": 0,
+                                    "y": 0,
+                                    "scaleX": 1,
+                                    "scaleY": 1,
+                                    "rotation": 0,
+                                },
+                                "volume": 1,
+                                "opacity": 1,
+                                "visible": True,
+                                "metadata": {},
+                            }
+                        ],
+                    }
+                ],
+                "transitions": [],
+                "markers": [],
+            },
+            "renderSettings": {"format": "mp4", "width": 1280, "height": 720, "frameRate": 30},
+        }
+    )
+
+
+async def actor(container, name: str):
+    key, secret = await container.api_keys.create(
+        APIKeyCreate(
+            name=name,
+            environment="test",
+            role=None,
+            scopes=[
+                "projects:read",
+                "projects:create",
+                "projects:update",
+                "jobs:create",
+                "jobs:cancel",
+            ],
+        ),
+        created_by="tests",
+    )
+    return key, await container.api_keys.authenticate(secret)
+
+
+@pytest.mark.asyncio
+async def test_editor_revision_isolation_render_idempotency_and_cancellation(
+    tmp_path: Path,
+) -> None:
+    container = build_container(settings(tmp_path))
+    await container.security_database.initialize()
+    try:
+        key_a, actor_a = await actor(container, "A")
+        _, actor_b = await actor(container, "B")
+        project = await container.projects.create(
+            workspace_id=actor_a.workspace_id,
+            user_id=actor_a.user_id,
+            api_key_id=key_a.id,
+            request_id=str(uuid4()),
+            payload=ProjectCreate(name="Studio"),
+        )
+        request_id = str(uuid4())
+        output = container.settings.output_dir / request_id
+        output.mkdir(parents=True)
+        source = output / "source.mp4"
+        source.write_bytes(b"test media")
+        asset = await container.assets.register_output(
+            workspace_id=actor_a.workspace_id,
+            user_id=actor_a.user_id,
+            request_id=request_id,
+            path=source,
+            mime_type="video/mp4",
+            project_id=project.id,
+        )
+        editor_document = document(project.id, asset.id)
+        saved = await container.editor.save(
+            workspace_id=actor_a.workspace_id,
+            user_id=actor_a.user_id,
+            api_key_id=key_a.id,
+            request_id=str(uuid4()),
+            project_id=project.id,
+            payload=EditorSaveRequest(expected_revision=0, schema_version=1, state=editor_document),
+        )
+        assert saved.revision == 1
+        with pytest.raises(ProjectEditorConflictError):
+            await container.editor.save(
+                workspace_id=actor_a.workspace_id,
+                user_id=actor_a.user_id,
+                api_key_id=key_a.id,
+                request_id=str(uuid4()),
+                project_id=project.id,
+                payload=EditorSaveRequest(
+                    expected_revision=0, schema_version=1, state=editor_document
+                ),
+            )
+        with pytest.raises(ProjectNotFoundError):
+            await container.editor.get(
+                workspace_id=actor_b.workspace_id,
+                user_id=actor_b.user_id,
+                project_id=project.id,
+            )
+        render_payload = ProjectRenderCreate(
+            editor_revision=1, output_format="mp4", width=1280, height=720
+        )
+        first = await container.renders.create(
+            workspace_id=actor_a.workspace_id,
+            user_id=actor_a.user_id,
+            api_key_id=key_a.id,
+            request_id=str(uuid4()),
+            project_id=project.id,
+            payload=render_payload,
+            idempotency_key="render-1",
+        )
+        second = await container.renders.create(
+            workspace_id=actor_a.workspace_id,
+            user_id=actor_a.user_id,
+            api_key_id=key_a.id,
+            request_id=str(uuid4()),
+            project_id=project.id,
+            payload=render_payload,
+            idempotency_key="render-1",
+        )
+        assert first.id == second.id and first.status == "queued"
+        with pytest.raises(ProjectRenderLimitError):
+            await container.renders.create(
+                workspace_id=actor_a.workspace_id,
+                user_id=actor_a.user_id,
+                api_key_id=key_a.id,
+                request_id=str(uuid4()),
+                project_id=project.id,
+                payload=render_payload,
+                idempotency_key="render-2",
+            )
+        cancelled = await container.renders.cancel(
+            workspace_id=actor_a.workspace_id,
+            user_id=actor_a.user_id,
+            api_key_id=key_a.id,
+            request_id=str(uuid4()),
+            project_id=project.id,
+            render_id=first.id,
+        )
+        assert cancelled.status == "cancelled"
+        repeated = await container.renders.cancel(
+            workspace_id=actor_a.workspace_id,
+            user_id=actor_a.user_id,
+            api_key_id=key_a.id,
+            request_id=str(uuid4()),
+            project_id=project.id,
+            render_id=first.id,
+        )
+        assert repeated.status == "cancelled"
+        async with container.security_database.tenant_session(
+            workspace_id=actor_a.workspace_id,
+            user_id=actor_a.user_id,
+        ) as session:
+            cancellation_events = list(
+                (
+                    await session.scalars(
+                        select(AuditEvent).where(
+                            AuditEvent.entity_id == first.id,
+                            AuditEvent.event_type == "project.render_cancelled",
+                        )
+                    )
+                ).all()
+            )
+        assert len(cancellation_events) == 1
+    finally:
+        await container.security_database.close()
+
+
+def test_render_compiler_is_deterministic_and_uses_server_paths(tmp_path: Path) -> None:
+    source = tmp_path / "source.mp4"
+    source.write_bytes(b"media")
+    state = document(str(uuid4()), str(uuid4()))
+    asset_id = next(iter(state.asset_ids()))
+    first = compile_render(
+        state,
+        asset_paths={asset_id: (source, "video/mp4")},
+        width=1280,
+        height=720,
+        frame_rate=30,
+        output_format="mp4",
+        quality="standard",
+        preset="balanced",
+    )
+    second = compile_render(
+        state,
+        asset_paths={asset_id: (source, "video/mp4")},
+        width=1280,
+        height=720,
+        frame_rate=30,
+        output_format="mp4",
+        quality="standard",
+        preset="balanced",
+    )
+    assert first == second
+    assert source in first.args
+    assert first.duration_ms == 1000
+    assert "yuv420p" in first.args
+
+
+def test_render_compiler_ignores_hidden_timeline_tail(tmp_path: Path) -> None:
+    source = tmp_path / "source.mp4"
+    source.write_bytes(b"media")
+    state = document(str(uuid4()), str(uuid4()))
+    payload = state.model_dump(by_alias=True)
+    hidden_track = deepcopy(payload["timeline"]["tracks"][0])
+    hidden_track.update({"id": "video-hidden", "name": "Hidden", "order": 1, "visible": False})
+    hidden_track["clips"][0].update(
+        {"id": "clip-hidden", "trackId": "video-hidden", "startMs": 120_000}
+    )
+    payload["timeline"]["tracks"].append(hidden_track)
+    state_with_hidden_tail = EditorDocument.model_validate(payload)
+    asset_id = next(iter(state_with_hidden_tail.asset_ids()))
+
+    plan = compile_render(
+        state_with_hidden_tail,
+        asset_paths={asset_id: (source, "video/mp4")},
+        width=1280,
+        height=720,
+        frame_rate=30,
+        output_format="webm",
+        quality="high",
+        preset="quality",
+    )
+
+    assert plan.duration_ms == 1000
+    assert plan.args.count(source) == 1
+    assert "18" in plan.args
+    assert "0" in plan.args

tests/test_cors.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import httpx
+
+from app.core.config import Settings
+from main import create_app
+
+
+async def test_configured_frontend_origin_receives_cors_headers(tmp_path: Path) -> None:
+    settings = Settings(
+        _env_file=None,
+        auth_enabled=False,
+        database_url=f"sqlite+aiosqlite:///{tmp_path / 'security.db'}",
+        temp_dir=tmp_path / "temp",
+        output_dir=tmp_path / "outputs",
+        cors_allowed_origins="https://workspace.example.vercel.app",
+    )
+    app = create_app(settings)
+    transport = httpx.ASGITransport(app=app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.options(
+            "/v1/projects",
+            headers={
+                "Origin": "https://workspace.example.vercel.app",
+                "Access-Control-Request-Method": "GET",
+                "Access-Control-Request-Headers": "Authorization",
+            },
+        )
+
+    assert response.status_code == 200
+    assert response.headers["access-control-allow-origin"] == (
+        "https://workspace.example.vercel.app"
+    )
+    assert "authorization" in response.headers["access-control-allow-headers"].lower()
+
+
+async def test_unconfigured_origin_receives_no_cors_authorization(tmp_path: Path) -> None:
+    settings = Settings(
+        _env_file=None,
+        auth_enabled=False,
+        database_url=f"sqlite+aiosqlite:///{tmp_path / 'security.db'}",
+        temp_dir=tmp_path / "temp",
+        output_dir=tmp_path / "outputs",
+        cors_allowed_origins="https://workspace.example.vercel.app",
+    )
+    app = create_app(settings)
+    transport = httpx.ASGITransport(app=app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.options(
+            "/v1/projects",
+            headers={
+                "Origin": "https://attacker.example",
+                "Access-Control-Request-Method": "GET",
+            },
+        )
+
+    assert response.status_code == 400
+    assert "access-control-allow-origin" not in response.headers

tests/test_database_migration_contracts.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import re
+import unittest
+from pathlib import Path
+
+
+MIGRATION_FILES = sorted(Path("app/projects/migrations").glob("*.sql"))
+ALLOWED_TABLES = {
+    "teams",
+    "team_members",
+    "invitations",
+    "project_collaborators",
+    "approval_workflows",
+    "approval_requests",
+    "review_comments",
+    "collaboration_activity",
+    "notification_preferences",
+}
+
+
+class MigrationContractTests(unittest.TestCase):
+    def test_latest_migration_adds_approval_request_workspace_integrity(self) -> None:
+        latest = MIGRATION_FILES[-1].read_text()
+
+        self.assertIn("alter table approval_requests", latest)
+        self.assertIn("add column if not exists workspace_id text", latest)
+        self.assertIn("alter column workspace_id set not null", latest)
+        self.assertIn("create index if not exists ix_approval_requests_workspace", latest)
+
+    def test_migrations_do_not_recreate_shared_tables(self) -> None:
+        counts: dict[str, int] = {}
+        for path in MIGRATION_FILES:
+            body = path.read_text()
+            for table in ALLOWED_TABLES:
+                counts[table] = counts.get(table, 0) + body.count(f"create table if not exists {table}")
+
+        for table, count in counts.items():
+            self.assertEqual(
+                count,
+                1,
+                f"Duplicate table creation detected for {table}: {count} migrations recreate it",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_database_url.py

@@ -0,0 +1,16 @@
+from app.core.database_url import normalize_async_database_url
+
+
+def test_bare_postgres_urls_use_the_installed_async_driver() -> None:
+    assert (
+        normalize_async_database_url("postgresql://user:secret@db.example/app")
+        == "postgresql+asyncpg://user:secret@db.example/app"
+    )
+    assert (
+        normalize_async_database_url("postgres://user:secret@db.example/app")
+        == "postgresql+asyncpg://user:secret@db.example/app"
+    )
+    assert (
+        normalize_async_database_url("postgresql+asyncpg://user:secret@db.example/app")
+        == "postgresql+asyncpg://user:secret@db.example/app"
+    )

tests/test_downloader.py

@@ -0,0 +1,20 @@
+from unittest.mock import AsyncMock
+
+import respx
+from httpx import Response
+
+from app.services.downloader import Downloader
+from app.services.validator import MediaValidator
+
+
+@respx.mock
+async def test_url_download_streams_to_disk(settings, tmp_path) -> None:
+    url = "https://media.example.test/sample.mp3"
+    respx.get(url).mock(
+        return_value=Response(200, content=b"ID3data", headers={"content-type": "audio/mpeg"})
+    )
+    downloader = Downloader(settings, MediaValidator(settings))
+    downloader.validate_url = AsyncMock()  # type: ignore[method-assign]
+    path, mime_type = await downloader.download(url, tmp_path)
+    assert path.read_bytes() == b"ID3data"
+    assert mime_type == "audio/mpeg"

tests/test_error_handling.py

@@ -0,0 +1,27 @@
+import pytest
+from fastapi.testclient import TestClient
+
+from app.core.exceptions import NotFoundError
+from main import create_app
+
+
+def test_errors_use_safe_standard_envelope(settings) -> None:
+    with TestClient(create_app(settings), raise_server_exceptions=False) as client:
+        response = client.post(
+            "/v1/probe",
+            json={"base64": "not-valid-base64!", "filename": "sample.mp3"},
+        )
+    assert response.status_code == 422
+    payload = response.json()
+    assert payload["success"] is False
+    assert payload["request_id"]
+    assert payload["error"]["code"] == "INVALID_INPUT"
+    assert "traceback" not in response.text.lower()
+
+
+def test_download_path_traversal_is_rejected(settings) -> None:
+    app = create_app(settings)
+    with pytest.raises(NotFoundError):
+        app.state.container.cleanup.resolve_download(
+            "00000000-0000-0000-0000-000000000000", "../secret.mp4"
+        )

tests/test_ffmpeg_operations.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import shutil
+import subprocess
+
+import pytest
+
+from app.models.media import InputMedia, MediaSource
+from app.operations.convert import convert_audio
+from app.services.ffmpeg_service import FFmpegService
+
+
+@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg is not installed")
+async def test_ffmpeg_audio_conversion(settings, tmp_path) -> None:
+    source = tmp_path / "tone.wav"
+    subprocess.run(
+        [
+            "ffmpeg",
+            "-hide_banner",
+            "-loglevel",
+            "error",
+            "-f",
+            "lavfi",
+            "-i",
+            "sine=frequency=440:duration=0.2",
+            "-y",
+            str(source),
+        ],
+        check=True,
+    )
+    media = InputMedia(
+        source=MediaSource.MULTIPART,
+        filename=source.name,
+        mime_type="audio/wav",
+        temp_path=source,
+        size=source.stat().st_size,
+    )
+    result = await convert_audio(
+        FFmpegService(settings), [media], {"format": "mp3"}, tmp_path / "out"
+    )
+    assert result.path is not None
+    assert result.path.is_file()
+    assert result.path.stat().st_size > 0
+
+
+async def test_ffmpeg_codec_listing_is_structured(settings, monkeypatch) -> None:
+    service = FFmpegService(settings)
+
+    async def fake_capture(*args, **kwargs) -> str:
+        return """Codecs:
+ D..... = Decoding supported
+ .E.... = Encoding supported
+ -------
+ DEV.LS h264 H.264 / AVC / MPEG-4 AVC
+ DEA.L. aac AAC (Advanced Audio Coding)
+"""
+
+    monkeypatch.setattr(service, "_capture", fake_capture)
+
+    codecs = await service.codecs()
+
+    assert [codec["name"] for codec in codecs] == ["h264", "aac"]
+    assert codecs[0]["decode"] is True
+    assert codecs[0]["encode"] is True
+    assert codecs[0]["type"] == "video"

tests/test_ffprobe.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+import shutil
+import wave
+
+import pytest
+
+from app.services.ffprobe_service import FFprobeService
+
+
+@pytest.mark.skipif(shutil.which("ffprobe") is None, reason="ffprobe is not installed")
+async def test_ffprobe_returns_audio_metadata(settings, tmp_path) -> None:
+    audio = tmp_path / "tone.wav"
+    with wave.open(str(audio), "wb") as stream:
+        stream.setnchannels(1)
+        stream.setsampwidth(2)
+        stream.setframerate(8000)
+        stream.writeframes(b"\x00\x00" * 8000)
+    metadata = await FFprobeService(settings).probe(audio)
+    assert metadata["duration"] == pytest.approx(1.0, abs=0.01)
+    assert metadata["audio_streams"][0]["codec"] == "pcm_s16le"

tests/test_generation_flux.py

@@ -0,0 +1,324 @@
+"""Mocked protocol tests for the audited FLUX.2 Klein worker integration."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from pathlib import Path
+from types import SimpleNamespace
+
+import httpx
+import pytest
+
+from app.core.config import Settings
+from app.generation.domain.enums import (
+    GenerationModality,
+    WorkerCancellationStatus,
+    WorkerErrorCategory,
+    WorkerJobStatus,
+)
+from app.generation.domain.errors import (
+    GenerationCapabilityUnsupportedError,
+    GenerationValidationError,
+    GenerationWorkerError,
+)
+from app.generation.domain.retry import GenerationRetryPolicy
+from app.generation.model_registry import GenerationModelRegistration, GenerationModelRegistry
+from app.generation.providers.flux import (
+    FLUX_BASE_MODEL_ID,
+    FLUX_DISTILLED_MODEL_ID,
+    FLUX_MODEL_CAPABILITY,
+    FLUX_MODEL_ID,
+    FLUX_PROVIDER_ID,
+    FluxProviderAdapter,
+)
+from app.generation.providers.worker_client import RemoteWorkerClient
+from app.generation.schemas.requests import GenerationRequestCreate
+
+
+async def _no_sleep() -> None:
+    return None
+
+
+def _client(
+    handler: Callable[[httpx.Request], httpx.Response], *, retries: int = 2
+) -> RemoteWorkerClient:
+    return RemoteWorkerClient(
+        base_url="https://flux-worker.example",
+        bearer_token="x" * 32,
+        connect_timeout_seconds=1,
+        request_timeout_seconds=1,
+        read_timeout_seconds=1,
+        retry_policy=GenerationRetryPolicy(max_retries=retries, backoff_seconds=0),
+        http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
+        sleep=lambda _: _no_sleep(),
+    )
+
+
+def _info() -> dict[str, object]:
+    return {
+        "id": FLUX_MODEL_ID,
+        "name": "FLUX.2 Klein 4B",
+        "type": "image",
+        "license": "Apache-2.0",
+        "status": "ready",
+        "models": {"distilled": FLUX_DISTILLED_MODEL_ID, "base": FLUX_BASE_MODEL_ID},
+    }
+
+
+def _payload(**overrides: object) -> GenerationRequestCreate:
+    value: dict[str, object] = {
+        "provider": FLUX_PROVIDER_ID,
+        "model_id": FLUX_MODEL_ID,
+        "modality": "image",
+        "prompt": "A cinematic coastal city at sunrise",
+        "flux": {
+            "mode_choice": "Distilled (4 steps)",
+            "seed": 42,
+            "randomize_seed": False,
+            "width": 1024,
+            "height": 1024,
+            "num_inference_steps": 4,
+            "guidance_scale": 1.0,
+            "prompt_upsampling": False,
+        },
+    }
+    value.update(overrides)
+    return GenerationRequestCreate.model_validate(value)
+
+
+@pytest.mark.asyncio
+async def test_flux_exact_model_discovery_and_readiness() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/health":
+            return httpx.Response(200, json={"status": "ok"})
+        if request.url.path == "/ready":
+            return httpx.Response(
+                200,
+                json={
+                    "status": "ready",
+                    "model_loaded": True,
+                    "model": FLUX_MODEL_ID,
+                    "accepting_jobs": True,
+                },
+            )
+        return httpx.Response(200, json=_info())
+
+    adapter = FluxProviderAdapter(client=_client(handler))
+    registry = GenerationModelRegistry(
+        [
+            GenerationModelRegistration(
+                provider_id=FLUX_PROVIDER_ID,
+                model=FLUX_MODEL_CAPABILITY,
+                configuration_reference="flux-space",
+            )
+        ]
+    )
+    assert (await adapter.health()).status.value == "healthy"
+    models = registry.verify_readiness(
+        provider_id=FLUX_PROVIDER_ID,
+        worker_info=await adapter.info(),
+        readiness=await adapter.ready(),
+        provider_configured=adapter.available,
+    )
+    assert models[0].model.id == FLUX_MODEL_ID
+    assert models[0].model.modality is GenerationModality.IMAGE
+    assert models[0].available
+
+
+@pytest.mark.asyncio
+async def test_flux_identity_mismatch_and_not_ready_are_not_advertised() -> None:
+    wrong = {**_info(), "models": {"distilled": "untrusted/model", "base": FLUX_BASE_MODEL_ID}}
+
+    def identity_handler(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json=wrong)
+
+    with pytest.raises(GenerationWorkerError) as raised:
+        await FluxProviderAdapter(client=_client(identity_handler)).info()
+    assert raised.value.category is WorkerErrorCategory.PROVIDER_ERROR
+
+    def not_ready_handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/ready":
+            return httpx.Response(
+                503,
+                json={
+                    "status": "not_ready",
+                    "model_loaded": False,
+                    "model": FLUX_MODEL_ID,
+                    "accepting_jobs": False,
+                },
+            )
+        return httpx.Response(200, json=_info())
+
+    with pytest.raises(GenerationWorkerError) as raised:
+        await FluxProviderAdapter(client=_client(not_ready_handler)).ready()
+    assert raised.value.category is WorkerErrorCategory.WORKER_NOT_READY
+
+
+@pytest.mark.asyncio
+async def test_flux_text_submission_uses_strict_form_and_has_no_automatic_retry() -> None:
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(202, json={"job_id": "flux_" + "a" * 32, "status": "queued"})
+
+    job = await FluxProviderAdapter(client=_client(handler)).submit(
+        payload={"prompt": "A city at sunrise", "flux": {"width": 1024, "height": 1024}},
+        idempotency_key="generation-request-id",
+    )
+    assert job.status is WorkerJobStatus.QUEUED
+    assert requests[0].headers["authorization"] == "Bearer " + "x" * 32
+    assert requests[0].headers["content-type"].startswith("application/x-www-form-urlencoded")
+    assert b"width=1024" in requests[0].content
+
+
+@pytest.mark.asyncio
+async def test_flux_optional_canonical_image_uses_multipart(tmp_path: Path) -> None:
+    source = tmp_path / "input.png"
+    source.write_bytes(b"image-input")
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        body = request.content.decode("latin-1")
+        assert 'name="input_images"' in body
+        assert 'name="prompt"' in body
+        return httpx.Response(202, json={"job_id": "flux_" + "b" * 32, "status": "queued"})
+
+    job = await FluxProviderAdapter(client=_client(handler)).submit(
+        payload={"prompt": "Edit this image"},
+        idempotency_key="generation-request-id",
+        input_path=source,
+        input_mime_type="image/png",
+    )
+    assert job.external_job_id.startswith("flux_")
+
+
+@pytest.mark.asyncio
+async def test_flux_rejects_invalid_requests_and_input_assets() -> None:
+    adapter = FluxProviderAdapter(client=None)
+    for invalid in ({"prompt": " "}, {"modality": "video"}):
+        with pytest.raises(Exception):
+            await adapter.validate_request(_payload(**invalid))
+
+    with pytest.raises(GenerationValidationError):
+        await adapter.validate_input_asset(
+            _payload(), SimpleNamespace(mime_type="video/mp4", file_size=100)
+        )
+    with pytest.raises(GenerationValidationError):
+        await adapter.validate_input_asset(
+            _payload(), SimpleNamespace(mime_type="image/png", file_size=21 * 1024 * 1024)
+        )
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("negative_prompt", "unsupported"),
+        ("scheduler", "unsupported"),
+        ("width", 1023),
+        ("height", 1032),
+    ],
+)
+def test_flux_schema_rejects_unsupported_or_invalid_parameters(field: str, value: object) -> None:
+    raw = _payload().model_dump()
+    flux = dict(raw["flux"] or {})
+    flux[field] = value
+    raw["flux"] = flux
+    with pytest.raises(ValueError):
+        GenerationRequestCreate.model_validate(raw)
+
+
+@pytest.mark.asyncio
+async def test_flux_rejects_controls_for_another_provider() -> None:
+    payload = _payload(wan={"duration_seconds": 1.0})
+    with pytest.raises(GenerationCapabilityUnsupportedError):
+        await FluxProviderAdapter(client=None).validate_request(payload)
+
+
+@pytest.mark.asyncio
+async def test_flux_completed_job_maps_a_safe_png_output_and_retrieves_it() -> None:
+    job_id = "flux_" + "c" * 32
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("/output"):
+            return httpx.Response(200, content=b"png-output")
+        return httpx.Response(
+            200,
+            json={
+                "job_id": job_id,
+                "status": "completed",
+                "output": {"type": "image", "filename": "output.png"},
+            },
+        )
+
+    adapter = FluxProviderAdapter(client=_client(handler))
+    job = await adapter.get_job(external_job_id=job_id)
+    assert job.output is not None
+    assert job.output.mime_type == "image/png"
+    assert job.output.download_path == f"/v1/jobs/{job_id}/output"
+    output = await adapter.retrieve_output(external_job_id=job_id)
+    async with adapter.stream_output(output) as chunks:
+        assert b"".join([chunk async for chunk in chunks]) == b"png-output"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status_code", [429, 502, 503, 504])
+async def test_flux_polling_uses_shared_bounded_transient_retry(status_code: int) -> None:
+    calls = 0
+    job_id = "flux_" + "d" * 32
+
+    def handler(_: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        if calls < 3:
+            return httpx.Response(status_code, json={"detail": {"token": "never-store"}})
+        return httpx.Response(200, json={"job_id": job_id, "status": "running"})
+
+    job = await FluxProviderAdapter(client=_client(handler, retries=2)).get_job(
+        external_job_id=job_id
+    )
+    assert job.status is WorkerJobStatus.RUNNING
+    assert calls == 3
+
+
+@pytest.mark.asyncio
+async def test_flux_permanent_error_is_not_retried_and_cancellation_is_accurate() -> None:
+    job_id = "flux_" + "e" * 32
+    calls = 0
+
+    def permanent_handler(_: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        return httpx.Response(400, json={"detail": {"code": "FLUX_REQUEST_INVALID"}})
+
+    with pytest.raises(GenerationWorkerError) as raised:
+        await FluxProviderAdapter(client=_client(permanent_handler, retries=3)).get_job(
+            external_job_id=job_id
+        )
+    assert raised.value.category is WorkerErrorCategory.INVALID_REQUEST
+    assert calls == 1
+
+    def queued_handler(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"job_id": job_id, "status": "cancelled"})
+
+    def running_handler(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            409,
+            json={"detail": {"code": "FLUX_JOB_NOT_CANCELLABLE", "status": "running"}},
+        )
+
+    assert (
+        await FluxProviderAdapter(client=_client(queued_handler)).cancel(external_job_id=job_id)
+    ).status is WorkerCancellationStatus.CANCELLED
+    assert (
+        await FluxProviderAdapter(client=_client(running_handler)).cancel(external_job_id=job_id)
+    ).status is WorkerCancellationStatus.FAILED
+
+
+def test_flux_configuration_is_optional_and_does_not_change_wan_configuration() -> None:
+    disabled = FluxProviderAdapter.from_settings(Settings(_env_file=None))
+    invalid = FluxProviderAdapter.from_settings(
+        Settings(_env_file=None, flux_space_url="https://flux-worker.example")
+    )
+    assert not disabled.available
+    assert not invalid.available
+    assert invalid.configuration_error is not None

tests/test_generation_foundation.py

@@ -0,0 +1,509 @@
+from __future__ import annotations
+
+import base64
+from contextlib import asynccontextmanager
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.container import build_container
+from app.ai.schemas import AiGenerateImageRequest
+from app.core.config import Settings
+from app.generation.domain.capabilities import (
+    GenerationModelCapability,
+    GenerationProviderCapabilities,
+)
+from app.generation.domain.enums import (
+    GenerationJobStatus,
+    GenerationModality,
+    WorkerCancellationStatus,
+    WorkerHealthStatus,
+    WorkerJobStatus,
+    WorkerReadinessStatus,
+)
+from app.generation.domain.errors import (
+    GenerationIdempotencyConflictError,
+    GenerationInputAssetNotFoundError,
+    GenerationJobNotFoundError,
+    GenerationProviderJobConflictError,
+)
+from app.generation.domain.runtime import (
+    WorkerCancellationResult,
+    WorkerHealth,
+    WorkerInfo,
+    WorkerJob,
+    WorkerOutput,
+    WorkerReadiness,
+)
+from app.generation.model_registry import (
+    GenerationModelRegistration,
+    GenerationModelRegistry,
+)
+from app.generation.providers.base import GenerationProviderAdapter
+from app.generation.providers.registry import GenerationProviderRegistry
+from app.generation.schemas.requests import GenerationRequestCreate
+from app.security.schemas import APIKeyCreate
+from main import create_app
+
+
+def generation_settings(tmp_path: Path) -> Settings:
+    return Settings(
+        _env_file=None,
+        auth_enabled=True,
+        database_url=f"sqlite+aiosqlite:///{tmp_path / 'security.db'}",
+        social_database_url=f"sqlite+aiosqlite:///{tmp_path / 'social.db'}",
+        social_auto_migrate=True,
+        social_worker_enabled=False,
+        social_oauth_encryption_key="test-only-encryption-material",
+        temp_dir=tmp_path / "temp",
+        output_dir=tmp_path / "outputs",
+        cleanup_interval_seconds=3600,
+        whisper_model="tiny",
+        generation_enabled=True,
+    )
+
+
+class AvailableTestProvider(GenerationProviderAdapter):
+    capabilities = GenerationProviderCapabilities(
+        provider="test-generation",
+        name="Test generation adapter",
+        implementation_status="test",
+        models=[
+            GenerationModelCapability(
+                id="test-image-v1",
+                name="Test image v1",
+                modality=GenerationModality.IMAGE,
+                input_asset_supported=True,
+            )
+        ],
+    )
+
+    def __init__(self) -> None:
+        self.cancellation_result = WorkerCancellationResult(
+            status=WorkerCancellationStatus.REQUESTED
+        )
+
+    @property
+    def available(self) -> bool:
+        return True
+
+    async def validate_request(self, payload: GenerationRequestCreate) -> dict[str, object]:
+        return {"prompt": payload.prompt}
+
+    async def health(self) -> WorkerHealth:
+        return WorkerHealth(status=WorkerHealthStatus.HEALTHY)
+
+    async def info(self) -> WorkerInfo:
+        return WorkerInfo(
+            id="test-generation-worker",
+            name="Test generation worker",
+            media_types=[GenerationModality.IMAGE],
+            models=[
+                {
+                    "id": "test-image-v1",
+                    "name": "Test image v1",
+                    "media_types": [GenerationModality.IMAGE],
+                }
+            ],
+            status=WorkerHealthStatus.HEALTHY,
+        )
+
+    async def ready(self) -> WorkerReadiness:
+        return WorkerReadiness(
+            status=WorkerReadinessStatus.READY,
+            model_loaded=True,
+            model_ids=["test-image-v1"],
+        )
+
+    async def cancel(self, *, external_job_id: str) -> WorkerCancellationResult:
+        assert external_job_id == "worker-job-1"
+        return self.cancellation_result
+
+    async def get_job(self, *, external_job_id: str) -> WorkerJob:
+        assert external_job_id == "worker-job-1"
+        return WorkerJob(
+            external_job_id=external_job_id,
+            status=WorkerJobStatus.COMPLETED,
+            output=WorkerOutput(
+                output_type=GenerationModality.IMAGE,
+                mime_type="image/png",
+                provider_output_id="worker-output-1",
+                download_path="/v1/outputs/worker-output-1",
+            ),
+        )
+
+    @asynccontextmanager
+    async def stream_output(self, output: WorkerOutput):
+        assert output.provider_output_id == "worker-output-1"
+
+        async def chunks():
+            yield base64.b64decode(
+                "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJ"
+                "AAAADUlEQVQIHWP4z8DwHwAFgAI/ScL9aQAAAABJRU5ErkJggg=="
+            )
+
+        yield chunks()
+
+
+async def create_context(container, name: str):
+    _, secret = await container.api_keys.create(
+        APIKeyCreate(
+            name=name,
+            environment="test",
+            role=None,
+            scopes=[
+                "generation:providers:read",
+                "generation:requests:read",
+                "generation:requests:create",
+                "generation:jobs:cancel",
+            ],
+        ),
+        created_by="tests",
+    )
+    return await container.api_keys.authenticate(secret)
+
+
+def request_payload(*, prompt: str = "A test image") -> GenerationRequestCreate:
+    return GenerationRequestCreate(
+        provider="test-generation",
+        model_id="test-image-v1",
+        modality=GenerationModality.IMAGE,
+        prompt=prompt,
+    )
+
+
+@pytest.fixture
+async def generation_container(tmp_path: Path):
+    container = build_container(generation_settings(tmp_path))
+    await container.security_database.initialize()
+    provider = AvailableTestProvider()
+    container.generation.providers = GenerationProviderRegistry([provider])
+    container.generation.models = GenerationModelRegistry(
+        [
+            GenerationModelRegistration(
+                provider_id=provider.provider,
+                model=provider.capabilities.models[0],
+                configuration_reference="test-generation-worker",
+            )
+        ]
+    )
+    await container.generation.initialize()
+    await container.generation.refresh_provider_runtime(provider.provider)
+    try:
+        yield container
+    finally:
+        await container.security_database.close()
+
+
+@pytest.mark.asyncio
+async def test_optional_generation_providers_start_unavailable_without_configuration(
+    tmp_path: Path,
+) -> None:
+    container = build_container(generation_settings(tmp_path))
+    await container.security_database.initialize()
+    await container.generation.initialize()
+    try:
+        providers = container.generation.list_providers()
+        assert [provider.capabilities.provider for provider in providers] == ["flux", "wan"]
+        assert not any(provider.available for provider in providers)
+        assert not container.generation.get_model("flux", "flux.2-klein-4b").available
+        assert not container.generation.get_model("wan", "wan2.2").available
+    finally:
+        await container.security_database.close()
+
+
+@pytest.mark.asyncio
+async def test_ai_studio_advertises_and_isolates_real_generation_history(
+    generation_container,
+) -> None:
+    context = await create_context(generation_container, "AI Studio")
+    capabilities = generation_container.ai.capabilities()
+    image_tool = next(tool for tool in capabilities.tools if tool.operation == "generate_image")
+    video_tool = next(tool for tool in capabilities.tools if tool.operation == "generate_video")
+    assert image_tool.available
+    assert not video_tool.available
+
+    ordinary = await generation_container.generation.create(
+        workspace_id=context.workspace_id,
+        user_id=context.user_id,
+        payload=request_payload(prompt="ordinary generation"),
+        idempotency_key="ordinary-generation-key",
+    )
+    ai_job = await generation_container.ai.create(
+        workspace_id=context.workspace_id,
+        user_id=context.user_id,
+        api_key_id=context.api_key_id,
+        request_id="ai-request",
+        payload=AiGenerateImageRequest(
+            operation="generate_image",
+            prompt="AI Studio generation",
+        ),
+        idempotency_key="ai-studio-generation-key",
+    )
+    history = await generation_container.ai.history(
+        workspace_id=context.workspace_id,
+        user_id=context.user_id,
+        offset=0,
+        limit=25,
+    )
+    assert [item.generation_id for item in history.items] == [ai_job.generation_id]
+    assert ordinary.id not in {item.generation_id for item in history.items}
+
+
+def test_application_starts_with_optional_providers_disabled_when_unconfigured(
+    tmp_path: Path,
+) -> None:
+    """No worker URL/token is needed merely to start the application."""
+
+    with TestClient(create_app(generation_settings(tmp_path))) as client:
+        providers = client.app.state.container.generation.list_providers()
+        assert [provider.capabilities.provider for provider in providers] == ["flux", "wan"]
+        models = client.app.state.container.generation.list_models()
+        assert [model.model.id for model in models] == ["flux.2-klein-4b", "wan2.2"]
+        assert not any(model.available for model in models)
+
+
+@pytest.mark.asyncio
+async def test_provider_discovery_requires_a_verified_model(generation_container) -> None:
+    """A configured adapter is not publicly usable before runtime verification."""
+
+    provider_id = "test-generation"
+    generation_container.generation.models.mark_unavailable(provider_id)
+    assert not generation_container.generation.get_provider(provider_id).available
+    assert not generation_container.generation.list_providers()[0].available
+
+    await generation_container.generation.refresh_provider_runtime(provider_id)
+    assert generation_container.generation.get_provider(provider_id).available
+
+
+@pytest.mark.asyncio
+async def test_generation_request_idempotency_and_cancel(generation_container) -> None:
+    context = await create_context(generation_container, "Generation A")
+    workspace_id = str(context.workspace_id)
+    user_id = str(context.user_id)
+
+    first = await generation_container.generation.create(
+        workspace_id=workspace_id,
+        user_id=user_id,
+        payload=request_payload(),
+        idempotency_key="generation-request-key",
+    )
+    replay = await generation_container.generation.create(
+        workspace_id=workspace_id,
+        user_id=user_id,
+        payload=request_payload(),
+        idempotency_key="generation-request-key",
+    )
+    assert replay.id == first.id
+    assert replay.job.id == first.job.id
+    with pytest.raises(GenerationIdempotencyConflictError):
+        await generation_container.generation.create(
+            workspace_id=workspace_id,
+            user_id=user_id,
+            payload=request_payload(prompt="Different request"),
+            idempotency_key="generation-request-key",
+        )
+
+    cancelled = await generation_container.generation.cancel(workspace_id, user_id, first.job.id)
+    assert cancelled.status is GenerationJobStatus.CANCELLED
+    retrieved = await generation_container.generation.get_request(workspace_id, user_id, first.id)
+    assert retrieved.status is GenerationJobStatus.CANCELLED
+
+
+@pytest.mark.asyncio
+async def test_generation_records_are_workspace_isolated(generation_container) -> None:
+    context_a = await create_context(generation_container, "Generation A")
+    context_b = await create_context(generation_container, "Generation B")
+    created = await generation_container.generation.create(
+        workspace_id=str(context_a.workspace_id),
+        user_id=str(context_a.user_id),
+        payload=request_payload(),
+        idempotency_key="generation-isolation-key",
+    )
+    with pytest.raises(GenerationJobNotFoundError):
+        await generation_container.generation.get_job(
+            str(context_b.workspace_id), str(context_b.user_id), created.job.id
+        )
+    assert (
+        await generation_container.generation.list_requests(
+            str(context_b.workspace_id), str(context_b.user_id)
+        )
+        == []
+    )
+
+
+@pytest.mark.asyncio
+async def test_generation_rejects_another_workspace_canonical_input_asset(
+    generation_container,
+) -> None:
+    context_a = await create_context(generation_container, "Generation A")
+    context_b = await create_context(generation_container, "Generation B")
+    request_id = "00000000-0000-0000-0000-000000000010"
+    output_dir = generation_container.settings.output_dir / request_id
+    output_dir.mkdir(parents=True)
+    output = output_dir / "owned-input.png"
+    output.write_bytes(b"canonical image")
+    asset = await generation_container.assets.register_output(
+        workspace_id=str(context_a.workspace_id),
+        user_id=str(context_a.user_id),
+        request_id=request_id,
+        path=output,
+        mime_type="image/png",
+    )
+    with pytest.raises(GenerationInputAssetNotFoundError):
+        await generation_container.generation.create(
+            workspace_id=str(context_b.workspace_id),
+            user_id=str(context_b.user_id),
+            payload=GenerationRequestCreate(
+                provider="test-generation",
+                model_id="test-image-v1",
+                modality=GenerationModality.IMAGE,
+                prompt="Use another workspace asset",
+                input_asset_id=asset.id,
+            ),
+            idempotency_key="generation-cross-asset-key",
+        )
+
+
+@pytest.mark.parametrize("forbidden_field", ["provider_payload", "worker_url", "output_url"])
+def test_generation_request_schema_rejects_client_supplied_provider_controls(
+    forbidden_field: str,
+) -> None:
+    payload: dict[str, object] = {
+        "provider": "test-generation",
+        "model_id": "test-image-v1",
+        "modality": "image",
+        "prompt": "A test image",
+    }
+    payload[forbidden_field] = {"unsafe": True}
+    with pytest.raises(ValueError):
+        GenerationRequestCreate.model_validate(payload)
+
+
+@pytest.mark.asyncio
+async def test_remote_cancellation_preserves_requested_and_confirmed_states(
+    generation_container,
+) -> None:
+    context = await create_context(generation_container, "Generation cancellation")
+    workspace_id = str(context.workspace_id)
+    user_id = str(context.user_id)
+    created = await generation_container.generation.create(
+        workspace_id=workspace_id,
+        user_id=user_id,
+        payload=request_payload(),
+        idempotency_key="generation-cancellation-key",
+    )
+    await generation_container.generation.repository.transition_job(
+        workspace_id,
+        created.job.id,
+        GenerationJobStatus.SUBMITTING,
+        user_id=user_id,
+    )
+    await generation_container.generation.bind_provider_job(
+        workspace_id=workspace_id,
+        user_id=user_id,
+        job_id=created.job.id,
+        worker_job_id="worker-job-1",
+    )
+    await generation_container.generation.repository.transition_job(
+        workspace_id,
+        created.job.id,
+        GenerationJobStatus.RUNNING,
+        user_id=user_id,
+    )
+
+    requested = await generation_container.generation.cancel(workspace_id, user_id, created.job.id)
+    assert requested.status is GenerationJobStatus.CANCEL_REQUESTED
+
+    provider = generation_container.generation.providers.get("test-generation")
+    assert isinstance(provider, AvailableTestProvider)
+    provider.cancellation_result = WorkerCancellationResult(
+        status=WorkerCancellationStatus.CANCELLED
+    )
+    confirmed = await generation_container.generation.cancel(workspace_id, user_id, created.job.id)
+    assert confirmed.status is GenerationJobStatus.CANCELLED
+
+
+@pytest.mark.asyncio
+async def test_provider_job_binding_and_output_ingestion_are_workspace_scoped(
+    generation_container,
+) -> None:
+    context_a = await create_context(generation_container, "Generation output A")
+    context_b = await create_context(generation_container, "Generation output B")
+    workspace_a, user_a = str(context_a.workspace_id), str(context_a.user_id)
+    workspace_b, user_b = str(context_b.workspace_id), str(context_b.user_id)
+    job_a = await generation_container.generation.create(
+        workspace_id=workspace_a,
+        user_id=user_a,
+        payload=request_payload(),
+        idempotency_key="generation-output-a",
+    )
+    job_b = await generation_container.generation.create(
+        workspace_id=workspace_b,
+        user_id=user_b,
+        payload=request_payload(),
+        idempotency_key="generation-output-b",
+    )
+    for workspace_id, user_id, job_id in (
+        (workspace_a, user_a, job_a.job.id),
+        (workspace_b, user_b, job_b.job.id),
+    ):
+        await generation_container.generation.repository.transition_job(
+            workspace_id,
+            job_id,
+            GenerationJobStatus.SUBMITTING,
+            user_id=user_id,
+        )
+
+    await generation_container.generation.bind_provider_job(
+        workspace_id=workspace_a,
+        user_id=user_a,
+        job_id=job_a.job.id,
+        worker_job_id="worker-job-1",
+    )
+    with pytest.raises(GenerationProviderJobConflictError):
+        await generation_container.generation.bind_provider_job(
+            workspace_id=workspace_b,
+            user_id=user_b,
+            job_id=job_b.job.id,
+            worker_job_id="worker-job-1",
+        )
+
+    await generation_container.generation.repository.transition_job(
+        workspace_a,
+        job_a.job.id,
+        GenerationJobStatus.RUNNING,
+        user_id=user_a,
+    )
+    completed = await generation_container.generation.ingest_completed_provider_output(
+        workspace_id=workspace_a,
+        user_id=user_a,
+        job_id=job_a.job.id,
+    )
+    assert completed.status is GenerationJobStatus.SUCCEEDED
+    assert completed.output_asset_id is not None
+    output_asset = await generation_container.assets.get_owned_by_id(
+        workspace_id=workspace_a,
+        user_id=user_a,
+        asset_id=completed.output_asset_id,
+    )
+    assert output_asset.mime_type == "image/png"
+    assert output_asset.metadata_json["generation"]["media"]["resolution"] == {
+        "width": 1,
+        "height": 1,
+    }
+    assert (
+        await generation_container.generation.ingest_completed_provider_output(
+            workspace_id=workspace_a,
+            user_id=user_a,
+            job_id=job_a.job.id,
+        )
+        == completed
+    )
+    with pytest.raises(GenerationJobNotFoundError):
+        await generation_container.generation.ingest_completed_provider_output(
+            workspace_id=workspace_b,
+            user_id=user_b,
+            job_id=job_a.job.id,
+        )

tests/test_generation_provider_runtime.py

@@ -0,0 +1,371 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+
+import httpx
+import pytest
+from pydantic import ValidationError
+
+from app.generation.domain.capabilities import (
+    GenerationModelCapability,
+    GenerationProviderCapabilities,
+)
+from app.generation.domain.enums import (
+    GenerationModality,
+    WorkerCancellationStatus,
+    WorkerErrorCategory,
+    WorkerHealthStatus,
+    WorkerReadinessStatus,
+)
+from app.generation.domain.errors import GenerationWorkerError
+from app.generation.domain.retry import GenerationRetryPolicy
+from app.generation.domain.runtime import WorkerInfo, WorkerOutput, WorkerReadiness
+from app.generation.model_registry import (
+    GenerationModelRegistration,
+    GenerationModelRegistry,
+)
+from app.generation.providers.base import GenerationProviderAdapter
+from app.generation.providers.registry import GenerationProviderRegistry
+from app.generation.providers.worker_client import RemoteWorkerClient
+
+
+def worker_client(
+    handler: Callable[[httpx.Request], httpx.Response] | None = None,
+    *,
+    retries: int = 2,
+    sleep_calls: list[float] | None = None,
+) -> RemoteWorkerClient:
+    async def sleep(delay: float) -> None:
+        if sleep_calls is not None:
+            sleep_calls.append(delay)
+
+    client = httpx.AsyncClient(
+        transport=httpx.MockTransport(
+            handler
+            or (lambda _: httpx.Response(200, json={"status": "ok"}))
+        )
+    )
+    return RemoteWorkerClient(
+        base_url="https://worker.example",
+        bearer_token="test-worker-token",
+        connect_timeout_seconds=1,
+        request_timeout_seconds=1,
+        read_timeout_seconds=1,
+        retry_policy=GenerationRetryPolicy(max_retries=retries, backoff_seconds=0),
+        http_client=client,
+        sleep=sleep,
+    )
+
+
+class RuntimeTestProvider(GenerationProviderAdapter):
+    capabilities = GenerationProviderCapabilities(
+        provider="runtime-test",
+        name="Runtime test provider",
+        implementation_status="test",
+        models=[
+            GenerationModelCapability(
+                id="runtime-image-v1",
+                name="Runtime image v1",
+                modality=GenerationModality.IMAGE,
+            )
+        ],
+    )
+
+
+def test_provider_and_model_registration_starts_unavailable() -> None:
+    provider = RuntimeTestProvider()
+    providers = GenerationProviderRegistry([provider])
+    assert providers.get("runtime-test") is provider
+    models = GenerationModelRegistry(
+        [
+            GenerationModelRegistration(
+                provider_id=provider.provider,
+                model=provider.capabilities.models[0],
+                configuration_reference="runtime-test-config",
+                metadata={
+                    "access_token": "must-not-survive",
+                    "diagnostic": (
+                        "Bearer must-not-survive "
+                        "https://worker.example/output?sig=secret"
+                    ),
+                    "download_url": "https://worker.example/output?sig=secret",
+                },
+            )
+        ]
+    )
+    view = models.get(provider.provider, "runtime-image-v1")
+    assert not view.available
+    assert "access_token" not in view.metadata
+    assert "download_url" not in view.metadata
+    assert "must-not-survive" not in str(view.metadata)
+
+
+def test_model_availability_requires_readiness_info_and_configuration() -> None:
+    model = GenerationModelCapability(
+        id="runtime-image-v1", name="Runtime", modality=GenerationModality.IMAGE
+    )
+    registry = GenerationModelRegistry(
+        [
+            GenerationModelRegistration(
+                provider_id="runtime-test",
+                model=model,
+                configuration_reference="runtime-test-config",
+            )
+        ]
+    )
+    info = WorkerInfo(
+        id="runtime-test-worker",
+        name="Runtime worker",
+        media_types=[GenerationModality.IMAGE],
+        models=[
+            {
+                "id": model.id,
+                "name": model.name,
+                "media_types": [GenerationModality.IMAGE],
+            }
+        ],
+    )
+    not_ready = WorkerReadiness(
+        status=WorkerReadinessStatus.STARTING,
+        model_loaded=False,
+        model_ids=[model.id],
+    )
+    assert not registry.verify_readiness(
+        provider_id="runtime-test",
+        worker_info=info,
+        readiness=not_ready,
+        provider_configured=True,
+    )[0].available
+    ready = WorkerReadiness(
+        status=WorkerReadinessStatus.READY, model_loaded=True, model_ids=[model.id]
+    )
+    assert registry.verify_readiness(
+        provider_id="runtime-test",
+        worker_info=info,
+        readiness=ready,
+        provider_configured=True,
+    )[0].available
+
+
+@pytest.mark.asyncio
+async def test_worker_health_readiness_info_and_bearer_authentication() -> None:
+    seen_headers: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen_headers.append(request.headers.get("authorization", ""))
+        if request.url.path == "/health":
+            return httpx.Response(200, json={"status": "ok"})
+        if request.url.path == "/ready":
+            return httpx.Response(
+                200,
+                json={"status": "ready", "model_loaded": True, "model": "model-v1"},
+            )
+        return httpx.Response(
+            200,
+            json={"id": "model-v1", "name": "Worker model", "type": "image", "status": "ready"},
+        )
+
+    client = worker_client(handler)
+    assert (await client.health()).status is WorkerHealthStatus.HEALTHY
+    readiness = await client.ready()
+    assert readiness.status is WorkerReadinessStatus.READY
+    assert readiness.model_ids == ["model-v1"]
+    info = await client.info()
+    assert info.media_types == [GenerationModality.IMAGE]
+    assert info.models[0].id == "model-v1"
+    assert seen_headers == ["Bearer test-worker-token"] * 3
+
+
+@pytest.mark.asyncio
+async def test_timeout_and_connection_failure_are_retryable_and_safe() -> None:
+    request = httpx.Request("GET", "https://worker.example/health")
+    for exception, category in (
+        (httpx.ReadTimeout("secret-token", request=request), WorkerErrorCategory.TIMEOUT),
+        (
+            httpx.ConnectError("Bearer test-worker-token", request=request),
+            WorkerErrorCategory.WORKER_UNAVAILABLE,
+        ),
+    ):
+        calls = 0
+
+        def handler(_: httpx.Request, error: Exception = exception) -> httpx.Response:
+            nonlocal calls
+            calls += 1
+            raise error
+
+        client = worker_client(handler, retries=1)
+        with pytest.raises(GenerationWorkerError) as raised:
+            await client.health()
+        assert raised.value.category is category
+        assert "test-worker-token" not in str(raised.value)
+        assert calls == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status_code", [429, 502, 503, 504])
+async def test_retryable_http_failures_use_bounded_retry(status_code: int) -> None:
+    calls = 0
+    delays: list[float] = []
+
+    def handler(_: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        if calls < 3:
+            return httpx.Response(status_code, json={"secret": "not surfaced"})
+        return httpx.Response(200, json={"status": "ok"})
+
+    client = worker_client(handler, retries=2, sleep_calls=delays)
+    assert (await client.health()).status is WorkerHealthStatus.HEALTHY
+    assert calls == 3
+    assert delays == [0, 0]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status_code", [400, 401])
+async def test_non_retryable_http_failures_do_not_retry(status_code: int) -> None:
+    calls = 0
+
+    def handler(_: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        return httpx.Response(status_code)
+
+    client = worker_client(handler, retries=3)
+    with pytest.raises(GenerationWorkerError) as raised:
+        await client.health()
+    assert calls == 1
+    assert raised.value.http_status == status_code
+
+
+@pytest.mark.asyncio
+async def test_unexpected_exception_is_not_automatically_retryable() -> None:
+    calls = 0
+
+    def handler(_: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        raise RuntimeError("programming failure with secret-token")
+
+    client = worker_client(handler, retries=3)
+    with pytest.raises(GenerationWorkerError) as raised:
+        await client.health()
+    assert raised.value.category is WorkerErrorCategory.UNKNOWN_ERROR
+    assert calls == 1
+    assert "secret-token" not in str(raised.value)
+
+
+@pytest.mark.asyncio
+async def test_worker_cancellation_and_output_contract() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "POST":
+            return httpx.Response(202, json={"status": "cancellation_requested"})
+        return httpx.Response(
+            200,
+            json={
+                "job_id": "job-1",
+                "status": "completed",
+                "output": {
+                    "type": "image",
+                    "mime_type": "image/png",
+                    "id": "output-1",
+                    "download_path": "/v1/outputs/output-1",
+                    "filename": "output.png",
+                },
+            },
+        )
+
+    client = worker_client(handler)
+    cancellation = await client.cancel("job-1")
+    assert cancellation.status is WorkerCancellationStatus.REQUESTED
+    output = await client.retrieve_output("job-1")
+    assert output.provider_output_id == "output-1"
+    assert output.download_path == "/v1/outputs/output-1"
+    with pytest.raises(ValidationError):
+        WorkerOutput(
+            output_type=GenerationModality.IMAGE,
+            mime_type="image/png",
+            provider_output_id="output-1",
+            download_path="https://attacker.example/output.png",
+        )
+    with pytest.raises(ValidationError):
+        WorkerOutput(
+            output_type=GenerationModality.IMAGE,
+            mime_type="image/png",
+            provider_output_id="output-1",
+            download_path="/v1/outputs/%2e%2e/secrets",
+        )
+
+
+@pytest.mark.asyncio
+async def test_empty_successful_cancellation_response_means_requested_not_cancelled() -> None:
+    client = worker_client(lambda _: httpx.Response(204))
+    result = await client.cancel("job-1")
+    assert result.status is WorkerCancellationStatus.REQUESTED
+
+
+@pytest.mark.asyncio
+async def test_output_stream_is_scoped_to_the_configured_worker_origin() -> None:
+    client = worker_client(lambda _: httpx.Response(200, content=b"worker-output"))
+    output = WorkerOutput(
+        output_type=GenerationModality.IMAGE,
+        mime_type="image/png",
+        provider_output_id="output-1",
+        download_path="/v1/outputs/output-1",
+    )
+    async with client.stream_output(output) as chunks:
+        received = b"".join([chunk async for chunk in chunks])
+    assert received == b"worker-output"
+
+
+@pytest.mark.asyncio
+async def test_worker_info_requires_a_discovered_model_match_for_availability() -> None:
+    model = GenerationModelCapability(
+        id="runtime-image-v1", name="Runtime", modality=GenerationModality.IMAGE
+    )
+    registry = GenerationModelRegistry(
+        [
+            GenerationModelRegistration(
+                provider_id="runtime-test",
+                model=model,
+                configuration_reference="runtime-test-config",
+            )
+        ]
+    )
+    readiness = WorkerReadiness(
+        status=WorkerReadinessStatus.READY, model_loaded=True, model_ids=[model.id]
+    )
+    undiscovered = WorkerInfo(
+        id="worker",
+        name="Worker",
+        media_types=[GenerationModality.IMAGE],
+        models=[{"id": "other-model", "name": "Other", "media_types": ["image"]}],
+    )
+    assert not registry.verify_readiness(
+        provider_id="runtime-test",
+        worker_info=undiscovered,
+        readiness=readiness,
+        provider_configured=True,
+    )[0].available
+
+
+def test_worker_url_and_path_validation_blocks_ssrf_and_traversal() -> None:
+    policy = GenerationRetryPolicy(max_retries=0, backoff_seconds=0)
+    for url in (
+        "http://example.com",
+        "https://10.0.0.1",
+        "http://169.254.169.254",
+        "https://169.254.169.254",
+        "https://worker.example/%2e%2e/internal",
+        "file:///etc/passwd",
+    ):
+        with pytest.raises(ValueError):
+            RemoteWorkerClient(
+                base_url=url,
+                bearer_token=None,
+                connect_timeout_seconds=1,
+                request_timeout_seconds=1,
+                read_timeout_seconds=1,
+                retry_policy=policy,
+            )
+    with pytest.raises(GenerationWorkerError):
+        RemoteWorkerClient._safe_external_id("job/../../metadata")

tests/test_generation_wan.py

@@ -0,0 +1,346 @@
+"""Mocked protocol tests for the audited WAN 2.2 worker integration."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from pathlib import Path
+
+import httpx
+import pytest
+
+from app.core.config import Settings
+from app.generation.domain.enums import (
+    GenerationModality,
+    WorkerCancellationStatus,
+    WorkerErrorCategory,
+    WorkerJobStatus,
+)
+from app.generation.domain.errors import GenerationWorkerError
+from app.generation.domain.retry import GenerationRetryPolicy
+from app.generation.model_registry import GenerationModelRegistration, GenerationModelRegistry
+from app.generation.providers.wan import (
+    WAN_MODEL_CAPABILITY,
+    WAN_MODEL_ID,
+    WAN_PROVIDER_ID,
+    WanProviderAdapter,
+)
+from app.generation.providers.worker_client import RemoteWorkerClient
+from app.generation.schemas.requests import GenerationRequestCreate
+
+
+def _client(
+    handler: Callable[[httpx.Request], httpx.Response], *, retries: int = 2
+) -> RemoteWorkerClient:
+    return RemoteWorkerClient(
+        base_url="https://wan-worker.example",
+        bearer_token="x" * 32,
+        connect_timeout_seconds=1,
+        request_timeout_seconds=1,
+        read_timeout_seconds=1,
+        retry_policy=GenerationRetryPolicy(max_retries=retries, backoff_seconds=0),
+        http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
+        sleep=lambda _: _no_sleep(),
+    )
+
+
+async def _no_sleep() -> None:
+    return None
+
+
+def _payload(**overrides: object) -> GenerationRequestCreate:
+    value: dict[str, object] = {
+        "provider": WAN_PROVIDER_ID,
+        "model_id": WAN_MODEL_ID,
+        "modality": "video",
+        "input_asset_id": "11111111-1111-4111-8111-111111111111",
+        "prompt": "Slow cinematic cloud movement",
+        "wan": {
+            "duration_seconds": 0.5,
+            "steps": 4,
+            "guidance_scale": 1.0,
+            "guidance_scale_2": 1.0,
+            "seed": 42,
+            "randomize_seed": False,
+        },
+    }
+    value.update(overrides)
+    return GenerationRequestCreate.model_validate(value)
+
+
+def _info() -> dict[str, object]:
+    return {
+        "id": "wan2.2",
+        "name": "WAN 2.2 FP8 AOTI Faster",
+        "type": "video",
+        "task": "image-to-video",
+        "status": "ready",
+        "model_id": "Wan-AI/Wan2.2-I2V-A14B-Diffusers",
+        "fps": 16,
+    }
+
+
+@pytest.mark.asyncio
+async def test_wan_exact_model_discovery_and_readiness() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/health":
+            return httpx.Response(200, json={"status": "ok", "service": "mediarouter-wan-worker"})
+        if request.url.path == "/ready":
+            return httpx.Response(
+                200,
+                json={
+                    "status": "ready",
+                    "model_loaded": True,
+                    "model": "wan2.2",
+                    "accepting_jobs": True,
+                },
+            )
+        return httpx.Response(200, json=_info())
+
+    adapter = WanProviderAdapter(client=_client(handler))
+    registry = GenerationModelRegistry(
+        [
+            GenerationModelRegistration(
+                provider_id=WAN_PROVIDER_ID,
+                model=WAN_MODEL_CAPABILITY,
+                configuration_reference="wan-space",
+            )
+        ]
+    )
+    models = registry.verify_readiness(
+        provider_id=WAN_PROVIDER_ID,
+        worker_info=await adapter.info(),
+        readiness=await adapter.ready(),
+        provider_configured=adapter.available,
+    )
+    assert models[0].model.id == WAN_MODEL_ID
+    assert models[0].model.modality is GenerationModality.VIDEO
+    assert models[0].available
+
+
+@pytest.mark.asyncio
+async def test_wan_not_ready_and_model_mismatch_are_not_advertised() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/ready":
+            return httpx.Response(
+                503,
+                json={
+                    "status": "not_ready",
+                    "model_loaded": False,
+                    "model": "other-model",
+                    "accepting_jobs": False,
+                },
+            )
+        return httpx.Response(
+            200, json={"status": "ok"} if request.url.path == "/health" else _info()
+        )
+
+    adapter = WanProviderAdapter(client=_client(handler))
+    with pytest.raises(GenerationWorkerError) as raised:
+        await adapter.ready()
+    assert raised.value.category is WorkerErrorCategory.WORKER_NOT_READY
+
+
+@pytest.mark.asyncio
+async def test_wan_model_identity_mismatch_remains_unavailable() -> None:
+    wrong_info = {
+        **_info(),
+        "id": "different-wan-model",
+        "name": "Different model",
+    }
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/ready":
+            return httpx.Response(
+                200,
+                json={
+                    "status": "ready",
+                    "model_loaded": True,
+                    "model": WAN_MODEL_ID,
+                    "accepting_jobs": True,
+                },
+            )
+        return httpx.Response(200, json=wrong_info)
+
+    adapter = WanProviderAdapter(client=_client(handler))
+    with pytest.raises(GenerationWorkerError) as raised:
+        await adapter.info()
+    assert raised.value.category is WorkerErrorCategory.PROVIDER_ERROR
+
+
+@pytest.mark.asyncio
+async def test_wan_submission_is_multipart_and_has_no_automatic_retry(tmp_path: Path) -> None:
+    seen: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request)
+        return httpx.Response(202, json={"job_id": "wan_" + "a" * 32, "status": "queued"})
+
+    source = tmp_path / "input.png"
+    source.write_bytes(b"not-decoded-in-adapter-test")
+    adapter = WanProviderAdapter(client=_client(handler))
+    job = await adapter.submit(
+        payload={"prompt": "slow movement", "wan": {"duration_seconds": 0.5, "steps": 4}},
+        idempotency_key="generation-request-id",
+        input_path=source,
+        input_mime_type="image/png",
+    )
+    assert job.status is WorkerJobStatus.QUEUED
+    assert job.external_job_id.startswith("wan_")
+    assert seen[0].headers["authorization"] == "Bearer " + "x" * 32
+    body = seen[0].content.decode("latin-1")
+    assert 'name="image"' in body
+    assert 'name="duration_seconds"' in body
+    assert 'name="width"' not in body
+
+
+@pytest.mark.asyncio
+async def test_wan_submission_connection_ambiguity_is_not_retried(tmp_path: Path) -> None:
+    calls = 0
+    request = httpx.Request("POST", "https://wan-worker.example/v1/generate")
+
+    def handler(_: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        raise httpx.ConnectError("Bearer " + "x" * 32, request=request)
+
+    source = tmp_path / "input.png"
+    source.write_bytes(b"input")
+    adapter = WanProviderAdapter(client=_client(handler, retries=3))
+    with pytest.raises(GenerationWorkerError) as raised:
+        await adapter.submit(
+            payload={"prompt": "slow movement"},
+            idempotency_key="generation-request-id",
+            input_path=source,
+            input_mime_type="image/png",
+        )
+    assert raised.value.category is WorkerErrorCategory.WORKER_UNAVAILABLE
+    assert calls == 1
+    assert "Bearer" not in str(raised.value)
+
+
+@pytest.mark.asyncio
+async def test_wan_completed_job_maps_a_safe_video_output() -> None:
+    job_id = "wan_" + "b" * 32
+
+    def handler(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            json={
+                "job_id": job_id,
+                "status": "completed",
+                "output": {"type": "video", "filename": f"{job_id}.mp4"},
+            },
+        )
+
+    adapter = WanProviderAdapter(client=_client(handler))
+    job = await adapter.get_job(external_job_id=job_id)
+    assert job.output is not None
+    assert job.output.mime_type == "video/mp4"
+    assert job.output.provider_output_id == job_id
+    assert job.output.download_path == f"/v1/jobs/{job_id}/output"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status_code", [429, 502, 503, 504])
+async def test_wan_polling_uses_shared_bounded_transient_retry(status_code: int) -> None:
+    calls = 0
+    job_id = "wan_" + "d" * 32
+
+    def handler(_: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        if calls < 3:
+            return httpx.Response(status_code, json={"detail": {"token": "never-store"}})
+        return httpx.Response(200, json={"job_id": job_id, "status": "running"})
+
+    job = await WanProviderAdapter(client=_client(handler, retries=2)).get_job(
+        external_job_id=job_id
+    )
+    assert job.status is WorkerJobStatus.RUNNING
+    assert calls == 3
+
+
+@pytest.mark.asyncio
+async def test_wan_polling_does_not_retry_permanent_client_errors() -> None:
+    calls = 0
+    job_id = "wan_" + "e" * 32
+
+    def handler(_: httpx.Request) -> httpx.Response:
+        nonlocal calls
+        calls += 1
+        return httpx.Response(400, json={"detail": {"code": "WAN_PARAMETERS_INVALID"}})
+
+    with pytest.raises(GenerationWorkerError) as raised:
+        await WanProviderAdapter(client=_client(handler, retries=3)).get_job(
+            external_job_id=job_id
+        )
+    assert raised.value.category is WorkerErrorCategory.INVALID_REQUEST
+    assert calls == 1
+
+
+@pytest.mark.asyncio
+async def test_wan_cancellation_only_confirms_queued_worker_cancellation() -> None:
+    def queued_handler(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"job_id": "wan_" + "c" * 32, "status": "cancelled"})
+
+    def running_handler(_: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            409,
+            json={
+                "detail": {
+                    "code": "WAN_JOB_NOT_CANCELLABLE",
+                    "message": "A running job cannot be cancelled.",
+                    "status": "running",
+                }
+            },
+        )
+
+    assert (
+        await WanProviderAdapter(client=_client(queued_handler)).cancel(
+            external_job_id="wan_" + "c" * 32
+        )
+    ).status is WorkerCancellationStatus.CANCELLED
+    assert (
+        await WanProviderAdapter(client=_client(running_handler)).cancel(
+            external_job_id="wan_" + "c" * 32
+        )
+    ).status is WorkerCancellationStatus.FAILED
+
+
+@pytest.mark.parametrize(
+    "invalid",
+    [
+        {"prompt": " "},
+        {"modality": "image"},
+        {"input_asset_id": None},
+    ],
+)
+@pytest.mark.asyncio
+async def test_wan_request_validation_rejects_invalid_required_values(
+    invalid: dict[str, object]
+) -> None:
+    adapter = WanProviderAdapter(client=None)
+    with pytest.raises(Exception):
+        payload = _payload(**invalid)
+        await adapter.validate_request(payload)
+
+
+@pytest.mark.parametrize("field", ["width", "height", "num_frames", "provider_payload"])
+def test_wan_schema_rejects_unsupported_parameters(field: str) -> None:
+    raw = _payload().model_dump()
+    wan = dict(raw["wan"] or {})
+    wan[field] = 1
+    raw["wan"] = wan
+    with pytest.raises(ValueError):
+        GenerationRequestCreate.model_validate(raw)
+
+
+def test_wan_configuration_is_optional_and_never_enables_flux() -> None:
+    disabled = WanProviderAdapter.from_settings(Settings(_env_file=None))
+    invalid = WanProviderAdapter.from_settings(
+        Settings(_env_file=None, wan_space_url="https://wan-worker.example")
+    )
+    assert not disabled.available
+    assert not invalid.available
+    assert invalid.configuration_error is not None
+    assert WAN_PROVIDER_ID == "wan"

tests/test_health.py

@@ -0,0 +1,13 @@
+from fastapi.testclient import TestClient
+
+from main import create_app
+
+
+def test_health_endpoint(settings) -> None:
+    with TestClient(create_app(settings)) as client:
+        response = client.get("/health")
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["success"] is True
+    assert payload["metadata"]["status"] == "healthy"
+    assert response.headers["x-request-id"] == payload["request_id"]

tests/test_input_resolver.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import base64
+from uuid import uuid4
+
+import pytest
+from starlette.requests import Request
+
+from app.container import build_container
+from app.core.exceptions import InputError
+from app.models.media import MediaSource
+
+
+def json_request(payload: bytes) -> Request:
+    sent = False
+
+    async def receive():
+        nonlocal sent
+        if sent:
+            return {"type": "http.disconnect"}
+        sent = True
+        return {"type": "http.request", "body": payload, "more_body": False}
+
+    request = Request(
+        {
+            "type": "http",
+            "method": "POST",
+            "path": "/v1/probe",
+            "headers": [(b"content-type", b"application/json")],
+            "query_string": b"",
+        },
+        receive,
+    )
+    request.state.request_id = str(uuid4())
+    return request
+
+
+async def test_resolves_json_base64(settings) -> None:
+    container = build_container(settings)
+    encoded = base64.b64encode(b"ID3-not-real-audio").decode()
+    request = json_request(
+        ('{"base64":"%s","filename":"sample.mp3","format":"wav"}' % encoded).encode()
+    )
+    resolved = await container.resolver.resolve(request)
+    assert resolved.primary.source is MediaSource.JSON_BASE64
+    assert resolved.primary.filename == "sample.mp3"
+    assert resolved.primary.temp_path.read_bytes() == b"ID3-not-real-audio"
+    assert resolved.params["filename"] == "sample.mp3"
+    assert resolved.params["format"] == "wav"
+
+
+async def test_resolves_n8n_binary_property(settings) -> None:
+    container = build_container(settings)
+    encoded = base64.b64encode(b"audio").decode()
+    payload = (
+        '{"binary":{"audio":{"data":"%s","fileName":"voice.mp3",'
+        '"mimeType":"audio/mpeg"}}}' % encoded
+    ).encode()
+    resolved = await container.resolver.resolve(json_request(payload))
+    assert resolved.primary.source is MediaSource.N8N_BINARY
+    assert resolved.primary.filename == "voice.mp3"
+    assert resolved.primary.temp_path.read_bytes() == b"audio"
+
+
+async def test_resolves_nested_template_input(settings) -> None:
+    container = build_container(settings)
+    encoded = base64.b64encode(b"RIFF-template-audio").decode()
+    request = json_request(
+        (
+            '{"template":"mp3","input":{"base64":"%s",'
+            '"filename":"source.wav","mime_type":"audio/wav"},"parameters":{}}' % encoded
+        ).encode()
+    )
+
+    resolved = await container.resolver.resolve(request)
+
+    assert resolved.primary.source is MediaSource.JSON_BASE64
+    assert resolved.primary.filename == "source.wav"
+    assert resolved.params["template"] == "mp3"
+    assert resolved.params["parameters"] == {}
+
+
+async def test_resolve_payload_copies_managed_temp_file(settings) -> None:
+    settings.output_dir.mkdir(parents=True)
+    source = settings.output_dir / "previous" / "clip.mp3"
+    source.parent.mkdir()
+    source.write_bytes(b"ID3-managed-media")
+    container = build_container(settings)
+
+    resolved = await container.resolver.resolve_payload({"temp_path": str(source)}, str(uuid4()))
+
+    assert resolved.primary.source is MediaSource.LOCAL_PATH
+    assert resolved.primary.temp_path != source
+    assert resolved.primary.temp_path.read_bytes() == source.read_bytes()
+
+
+async def test_resolve_payload_rejects_unmanaged_path(settings, tmp_path) -> None:
+    source = tmp_path / "outside.mp3"
+    source.write_bytes(b"ID3-unmanaged-media")
+    container = build_container(settings)
+
+    with pytest.raises(InputError, match="TEMP_DIR or OUTPUT_DIR"):
+        await container.resolver.resolve_payload({"temp_path": str(source)}, str(uuid4()))